让Agent真正跑起来很简单但让它持续稳定地跑上几个小时、几天甚至几周完全是另一回事。我最近在做一个数据巡检类的Agent刚开始任务跑几分钟就结束一切正常一旦改成长期驻留、按小时自动执行各种奇怪的问题全冒出来了——上下文越堆越乱、跑着跑着突然停住不响应、同样的任务被重复执行了好几遍。这篇文章就把我在Long-Running Tasks这个方向上踩过的坑和最终沉淀下来的实现方案做个梳理给同样在折腾Agent长时运行的朋友一份可以直接参考的工程笔记。这个主题的难点不在于“怎么调大模型”而在于“怎么让一个有状态的执行体系在不可靠的组件之上保持可靠”。你会面对窗口限制、状态丢失、接口超时、重复执行、资源泄漏等一系列问题而且这些问题往往不是单个出现而是连环爆发。先别急着写代码把问题拆清楚比什么都重要。1. 长期运行Agent到底难在哪先看清问题全貌1.1 你以为的“运行中”和实际发生的“运行中”短任务场景里Agent的典型形态是“接收一个请求调用几次模型返回结果”整个过程秒级到分钟级。这时代理进程只有一个动作等模型回复。即使中途出问题重启整个进程重跑一次就行代价很低。但长期运行任务完全不同。一个Agent一旦进入“长跑模式”它的生命周期不再是一次请求而是一个持续执行的过程从外部系统拉数据、做判断、调工具、写结果然后进入下一个循环。这个过程中Agent不仅要记住它已经做了什么、下一步要做什么还要应对运行期间不断变化的外部环境——接口变慢了、目标服务重启了、数据格式改了。我在实际项目里最直观的感受是短任务是在“跑一段逻辑”长任务是在“维护一个状态机”。如果你还用短任务的心态去做长任务代码里所有状态都堆在内存变量里那系统崩溃的那一刻整个Agent的“记忆”就被格式化了一切从头再来。1.2 四类典型长期运行场景场景类型典型特征运行周期对Agent的要求批处理式长任务一次性下发大量数据逐个处理分钟到小时级可断点续跑、可跳过失败项持续监听式任务监听消息队列或事件流触发后处理天级驻留高可用、消息幂等、自动恢复人机协同审批流需要人工审核环节审核通过后继续小时到天级状态挂起、恢复唤醒、超时提醒定时巡检型Agent按固定周期执行检查并生成报告长期循环上下文隔离、结果归档、防重入这个分类不是为了分类而分类而是因为每种场景对“长期运行”的诉求完全不同。批处理式长任务的核心是“进度不能白费”所以你必须做检查点监听式任务的核心是“消息不能重复处理”所以你必须做幂等和偏移量管理人机协同的核心是“等待期间不能把状态丢了”所以你必须把状态持久化到外部存储而定时巡检的核心是“每次执行之间不能互相污染”所以你必须做会话隔离和结果归档。我见过很多人在架构上照搬别人的方案结果发现场景根本对不上。比如给一个批处理任务加了消息队列和复杂的恢复机制但实际崩溃一次重跑的成本也很低完全没必要。先定义清楚你的场景属于哪一类再决定要投入多少工程成本顺序不能反。1.3 长期运行的三大底层矛盾矛盾一有界上下文 vs 无限任务。模型输入有限制这是硬约束。长期运行的Agent会不断产生新的信息本轮执行结果、工具返回值、用户新增的指令。如果全部塞进上下文很快就会超出窗口限制。就算窗口足够大塞满旧内容也会大幅增加推理延迟和费用而且模型对中段内容的注意力会明显衰减出现“记忆幻觉”——它以为它记得实际上已经忘了。矛盾二单次执行模型 vs 持续执行需求。目前主流大模型的设计仍然是“单轮对话/单次推理”范式模型本身没有“持续运行”的概念也没有内置“等一下再继续”的机制。Agent的持续性完全要靠外层代码来维持你要负责调用模型、解析结果、保存中间状态、安排下一步调度。这等于说模型的不可靠性必须由工程层来兜底这是很多纯算法背景的人容易忽略的。矛盾三外部依赖的不可靠 vs 交付结果的确定性。Agent跑得越久依赖的外部组件就越多模型接口、工具服务、数据库、消息队列。任何一个组件抖动都会影响Agent的正常运行。如果Agent在设计上没有把“失败”当成一等公民来处理那它就会在某个深夜因为一次网络抖动卡在某个中间步骤然后永远停在那个状态。一句话总结**长期运行Agent的核心挑战不是“怎么让模型更聪明”而是“怎么让系统在模型和外部组件都不可靠的情况下依然可以恢复、可以继续、可以交付”。**理解了这三组矛盾后面所有的方案逻辑都会自然浮出来。2. 关键方案拆解怎么设计一个不容易死的Agent2.1 状态外置把“记忆”搬出聊天窗口长期运行Agent的第一个设计原则就是任何重要的状态都不能只存在内存里更不能只存在于模型的上下文窗口里。我在一开始设计那个数据巡检Agent的时候把所有执行记录都堆在系统提示词后面让模型“记住”自己跑到哪了。结果跑到第10轮左右就开始出问题输出格式开始漂移偶尔会漏掉前面的关键步骤有时候还会重复执行已经完成的操作。原因很简单上下文越长模型对早期内容的注意力越低而且Token开销每轮都在增加。后来我把所有状态全部外置到存储里。具体做了两件事第一设计一个简单的StateStore接口把Agent的执行状态塞进去。最开始我用的是SQLite后来换成PostgreSQL接口不变。核心字段就三个agent_id、current_step、context_snapshot。第二每一轮执行开始前从存储里恢复状态每一轮执行结束后把新状态写回去。整个过程和传统服务端开发里的“无状态化改造”几乎一模一样。class StateStore: def save(self, agent_id: str, step: str, payload: dict): # 将执行状态持久化payload可以是结构化数据快照 raise NotImplementedError def load(self, agent_id: str): # 恢复该Agent的最新状态 raise NotImplementedError把状态外置带来的另外一个好处是**你可以随时把Agent搬运到另一台机器上继续跑只要存储是共享的。**这对于后面做故障转移非常关键。所谓“长期运行”本质上就是状态不断被保存、加载、更新、再保存的循环而不是进程不重启。2.2 三步检查点机制崩溃后从哪接着跑状态外置解决了“状态在哪”的问题但还没解决“状态要存到什么程度”的问题。如果每执行一个原子步骤就全量快照一次磁盘IO会很夸张而且大部分快照内容是重复的。我实践下来效率比较高的是“事件溯源定期快照”的组合。具体做法分三步第一步定义事件。不要存“当前状态是什么”而是存“发生了什么”。比如一个事件叫task_started一个叫tool_call_succeeded一个叫step_completed。每个事件带上agent_id、事件类型、时间戳、以及本次事件的必要负载。第二步追加写事件日志。每发生一件事就往事件表里插一条记录。这个操作天然是追加式的顺序写性能很好。事件日志本身就是一个完整的审计轨迹排错的时候直接重放日志就能还原Agent当时的决策过程。第三步定期做快照。每隔N个事件或者每隔一定时间把当前完整状态压缩成一份快照存起来。恢复的时候找到最近一份快照加载然后重放快照之后的事件日志把状态推进到最新。class EventStore: def append(self, event: dict): # 追加写入事件事件是不可变的事实记录 pass def replay(self, agent_id: str, since_event_id: int): # 重放某个Agent在指定事件之后的所有事件 pass class AgentRuntime: def recover(self, agent_id: str): state self.snapshot_store.load_latest(agent_id) if state is None: state self.initial_state() events self.event_store.replay( agent_id, since_event_idstate.get(event_id, 0) ) for event in events: state self.apply_event(state, event) return state用事件溯源还有个附加福利排查线上问题特别爽。以前出问题只能看日志猜现在直接把执行轨迹拉出来每一步都有据可查。检查点打点的时机也很关键。我踩过一个坑在模型调用返回之后才做快照一旦模型调用之前工具已经产生了副作用比如发送了邮件快照恢复后重复执行会导致副作用重复发生。所以后来的原则是有副作用的操作执行前记录“意图事件”执行成功后记录“完成事件”快照以“完成事件”为恢复边界。2.3 任务队列与重试让Agent学会“等待”长期运行意味着Agent免不了要面对“此刻调不了模型”“此刻目标服务限流”“此刻依赖组件挂了”这些情况。这时候硬等是最差的选择会白白占用线程资源而是应该把当前任务放回队列让调度器稍后重新调度。我通常会用两种形态的队列一种是外部消息队列比如RabbitMQ或Kafka。这种适合批处理和监听场景任务从队列里取出来处理如果失败就投递回队列等待重试。消息队列自带持久化重启不丢消息天然适合长周期任务。另一种是内部重试队列用简单的Timer或DelayedQueue实现。这种适合步骤少、延迟需求不高的场景。比如在调用模型接口超时后我不会立刻重试而是先按指数退避计算延迟再塞回内部队列。关于重试我的两个核心经验经验一必须处理退避不能无脑重试。很多模型接口限流是以秒或分钟为窗口的立即重试只会加重限流。我是按“1秒、2秒、4秒、8秒、30秒、2分钟、10分钟”的阶梯做指数退避最大重试次数控制在5到7次。超过最大次数后把任务标记为needs_human_intervention并通知管理员。经验二重试必须有幂等键。如果不带幂等键重试时同一个任务可能被执行两次。我统一在任务入队时生成一个全局唯一的task_id所有下游处理都以这个task_id做去重和判断。这个task_id也是我前面事件日志里的关联ID整个链路都能串起来。import time import random def retry_with_backoff(func, max_retries5, base_delay1): for attempt in range(max_retries): try: return func() except TransientError: if attempt max_retries - 1: raise delay base_delay * (2 ** attempt) random.uniform(0, 0.5) time.sleep(delay)2.4 心跳与看门狗没有人盯着它就跑偏进程不退出不代表Agent真的活着。更准确地说Agent可能卡在某个模型调用上或者陷入一个无人知晓的死循环。此时需要一个外部观察者来确认“Agent还活着且还在推进”。我的实现是在Agent主循环里定期写入心跳记录更新最后活跃时间。然后单独起一个Watchdog进程每隔固定时间扫描心跳表找出那些“最后活跃时间超过阈值”的Agent实例判定为卡死然后触发恢复流程。心跳这里有个细节容易被忽略判断阈值不能设太短。有些Agent的循环会包含长时间的外部等待比如人工审批可能要等几小时。如果阈值太短会把正常的等待误判为卡死触发不必要的重启反而打断流程。我的做法是把心跳分为两类一类是“进程级心跳”证明Agent进程还在运行每30秒更新一次另一类是“步骤级心跳”证明Agent还在推进具体步骤只在步骤切换时更新。人工等待期间不更新步骤级心跳但进程级心跳始终在写。看门狗优先检查进程级心跳辅以步骤级心跳判断是否卡在某个步骤上。def watchdog_scan(): stuck find_agents( last_process_heartbeat_beforenow() - 90, statusrunning ) for agent in stuck: force_restart(agent.agent_id)这里的逻辑用一句大白话表达就是你自己说你还活着不算数得有另一个进程每隔一会儿来摸摸你的脉。有了心跳和看门狗Agent才算真正有了“被托管”的感觉。3. 实操落地的完整流程从架构选型到运行观察3.1 选型状态机编排 vs 纯自然语言驱动解决完基础机制问题后下一个问题是Agent的执行流程到底怎么组织我在这个点上试过两种路线差异非常大。第一种是纯自然语言驱动让模型自己决定下一步做什么。好处是灵活坏处是完全不可控。长期跑下来模型可能会自己发明一些流程跳过必要的校验步骤。如果是自由探索类的Agent比如帮你调研资料问题不大但如果是偏执行的Agent比如自动审批、数据订正这种自由度就是风险。第二种是状态机编排。先把整个流程的步骤定义清楚每个步骤明确输入、动作、输出、下一个状态的跳转条件。Agent在做完每步后根据结果决定跳转到哪个状态。模型只在某个具体的状态节点上负责“生成内容”或“做决策”而不是主导整个流程。我的最终选择是混合模式关键路径用状态机自由探索用贴近自然语言的子流程。举个例子在那个数据巡检Agent里整体流程是状态机每个状态节点内部模型可以自由调用工具、生成分析结果但流程的跳转控制权始终在代码手里。这样做的好处非常明显流程的可预测性大幅提升长期运行也不会发生“流程漂移”。每次出问题时我只需要看当前处于哪个状态、什么条件导致跳转失败而不是面对一团模型自然生成的“自由意志”。3.2 核心代码骨架一个可持久化的状态机Agent我写了一个简化版的状态机Agent骨架可以清楚表达整个机制是怎么配合工作的。from enum import Enum from dataclasses import dataclass class AgentState(str, Enum): INIT INIT FETCHING FETCHING ANALYZING ANALYZING REPORTING REPORTING DONE DONE FAILED FAILED dataclass class AgentContext: agent_id: str state: AgentState payload: dict event_id: int核心循环就是这样一段代码class PersistentAgentLoop: def __init__(self, state_store, event_store, tool_registry): self.state_store state_store self.event_store event_store self.tool_registry tool_registry def run(self, agent_id: str): ctx self.state_store.load(agent_id) while ctx.state ! AgentState.DONE and ctx.state ! AgentState.FAILED: self._heartbeat(agent_id) new_ctx self._step(ctx) self.state_store.save(agent_id, new_ctx) ctx new_ctx def _step(self, ctx): event {agent_id: ctx.agent_id, event_id: ctx.event_id 1} if ctx.state AgentState.INIT: return self._on_init(ctx, event) elif ctx.state AgentState.FETCHING: return self._on_fetching(ctx, event) elif ctx.state AgentState.ANALYZING: return self._on_analyzing(ctx, event) elif ctx.state AgentState.REPORTING: return self._on_reporting(ctx, event) raise UnexpectedStateError(ctx.state) def _on_fetching(self, ctx, event): # 调用工具抓取数据成功后写一个事件 result self.tool_registry.invoke(fetch_data, ctx.payload) self.event_store.append({**event, type: fetch_succeeded, result: result}) return AgentContext( agent_idctx.agent_id, stateAgentState.ANALYZING, payload{fetch_result: result}, event_idctx.event_id 1, )这个骨架的关键点在于每次循环都会把状态保存回存储崩溃后重新加载就是最后一步执行前的状态。配合事件日志连“最后一步执行完了但没有保存新状态”这种边缘情况也能在事件重放中还原出来。3.3 运行观察日志、追踪与告警长期运行的Agent可观测性直接决定事故恢复速度。我的经验是Agent项目的日志不能只按“INFO/WARN/ERROR”分级而是要按“执行链”来组织。每一条日志都要带上agent_id、event_id、组件名保证你可以按ID拉出一整条执行轨迹。我目前用一个轻量方案把结构化日志写到一段时序数据库里同时写出一个trace_id。在这个trace_id下的所有日志——包括模型请求、工具调用、状态切换、checkpoint保存——都能一次性查询出来。这比传统全局检索日志高效得多。# 查看某个Agent实例的完整执行轨迹 SELECT * FROM agent_events WHERE agent_id agent_001 ORDER BY event_id;告警方面我目前只设了四个触发器连续N次重试仍然失败需要人工介入。心跳超时Agent被判定为卡死。恢复流程触发了超过两次。某一步的执行时长超过预估值的三倍。总结一句话让长期运行的Agent“透明化”你才敢放心让它跑着不管。透明化程度越高你对系统的掌控感越强。4. 长期运行过程中的典型翻车现场4.1 死循环自我纠错成了无限转圈长期运行Agent最典型的问题就是模型陷入自我纠错的死循环。比如它调一个工具失败了于是它想“我换个方式试试”结果连续试了十几次全是同一类参数只是表述不同。从模型的角度看它每一轮都“试了新方案”但从工程的角度看它原地踏步了半小时。我当时给巡检Agent加了一个“循环检测”机制统计最近N条事件如果在连续的K次事件里状态始终没有发生有效变化比如工具名相同、输入参数高度相似就强制中断循环标记为stuck_in_loop等待人工处理。这个机制在自然语言驱动的Agent里尤其重要因为自然语言驱动下模型很难意识到自己已经在重复劳动。4.2 状态漂移存了状态但恢复后行为变了另一个更隐蔽的问题即使你正确保存和恢复了状态Agent恢复后的行为也可能发生变化。原因在于模型的输出受温度参数、随机种子和prompt变化的影响。比如你某天优化了系统提示词或者改了某个工具的描述那么之前快照下的Agent决策链路在重放后可能得到完全不同的结果。这不是持久化机制的错而是底层模型本身的非确定性。我的做法是在每个版本的快照里记录prompt版本号和工具schema版本号。恢复时检查版本号如果不一致就触发“轻量重规划”——让模型基于当前版本的工具和prompt重新生成后续步骤而不是硬着头皮重放旧事件。这个方案相当于给了Agent一个“模式切换”的能力在地基变了之后主动调整策略而不是死守旧地图。4.3 内存与资源泄漏跑两天后悄悄死了这类问题不发生在Agent逻辑本身而发生在宿主进程上。典型的泄漏源有会话历史对象无限增长、线程池没有关闭、HTTP客户端连接没有释放、外部调用的响应对象占着内存不回收。我分享一个很直接的经验长期运行的Agent进程必须做内存监控并且要在项目初期就预留“重启即恢复”的通道。合同约定地再完美的代码也挡不住某个第三方库的泄漏与其和泄漏搏斗不如把“优雅重启”做成一个一等公民功能。我现在的方案是当进程内存超过阈值时自动触发一次“生命周期结束”流程——保存状态、关闭连接、平滑退出然后由调度器拉一个新的进程接管。4.4 并发与幂等两个实例同时处理同一个任务如果长期运行Agent部署了多个副本并发问题就一定会出现。比如看门狗误判了一个Agent卡死触发了恢复流程在老Agent还没完全退出时就拉起了新实例结果两个实例都在处理同一个任务。针对这种情况我做了两个层面的兜底第一层是“实例锁”用数据库行锁实现。每个Agent在启动时都尝试获取锁只有拿到锁的实例才真正执行循环。第二层是“幂等消费者”在消息消费侧做去重。每条任务消息带task_id处理前检查处理记录表如果已经处理过就跳过。不依赖消息队列的“至少一次”还是“至多一次”语义而是自己保证“有效一次”。INSERT INTO task_dedup (task_id, agent_id, created_at) VALUES (?, ?, ?) ON CONFLICT (task_id) DO NOTHING; -- 影响行数为0说明任务已处理过4.5 大模型接口故障与限流最后聊聊外部依赖这是长期运行中遇到频率最高的故障源。我的经验总结成了一个简单的判定策略先区分错误类型再决定应对方式。rate_limit错误退避重试最大重试次数可放宽到8次。timeout错误退避重试但需要同步检查目标服务健康状态。invalid_request错误不重试直接标记为失败说明是入参或Agent自身的问题。server_error错误等待更长时间后重试大概率是服务方临时故障。还有一个很实用的技巧在持续失败时为Agent准备一条“降级路径”。比如主模型接口挂了切到备用模型模型全挂了退化为规则逻辑——至少保证完成核心动作而不是完全停摆。5. 一点经验总结想做长期运行先把这些原则刻在脑子里如果让我把长期运行Agent的经验浓缩成几条“早知道就好”的建议大概是这几条第一从第一天就假定进程会随时崩溃把所有状态持久化。不要想着“等系统稳定了再做”等你在生产环境里抓崩溃现场的时候就已经付出代价了。第二把所有有副作用的操作都记录成事件并且让事件成为系统里最高优先级的“事实来源”。你永远不会为“多记录了一条日志”而后悔但一定会为“少记录了一件事”而抓狂。第三给Agent一条“主动说我不行”的通道。我见过太多Agent在模型反复出错时硬撑最后用光所有预算和时间。设置好最大重试次数、最大循环检测次数以及“转人工”的信号比什么都管用。第四进程不退出不等于Agent活着。心跳、看门狗、执行链日志这三样是长期运行系统的基本盘一个都不能少。最后再分享一个小经验如果你刚开始改造一个Agent来支撑长期任务不要一上来就上Kafka、上分布式锁、上K8s。先用数据库加消息队列把核心机制搭起来跑通一个端到端的任务再逐步把瓶颈点在监控数据的引导下去优化。我现在的这套体系核心也就是一个PostgreSQL、一个任务队列再加一个监听进程但已经能支撑起数据巡检Agent连续跑上几个礼拜不崩。复杂度是要靠真实压力逼出来的不是靠设计阶段凭空堆出来的。