1. 从一次任务中断说起Agent循环执行的真实痛点凌晨两点我盯着日志里那行agent execution terminated due to error发呆。一个跑了四十多分钟的自动化任务在第37步调用外部接口时因为网络抖动直接挂掉前面36步积累的上下文、中间产物、工具调用记录全部归零。第二天重跑又卡在同一个位置。这种场景做过Agent开发的人应该都不陌生——我们花了大量精力在提示词工程、工具编排、模型选型上却常常忽略一个更底层的问题Agent的运行机制本身是否健壮。这篇内容想聊的就是这件事。标题里的几个词——上下文、检查点、任务恢复、循环执行、资源管控——看起来像是五个独立话题实际上它们是一条链上的五个环节。一个能真正跑在生产环境里的Agent不是能调用工具、能返回结果就完事了而是要在循环执行的过程中持续维护上下文在关键节点写入检查点在异常发生后能恢复任务同时全程做好资源管控。缺任何一环Agent都只能停留在Demo阶段。适合谁看如果你正在做Agent开发已经跑通过简单的工具调用链路但一遇到长任务、多轮循环、外部依赖不稳定就出问题那这篇内容就是给你准备的。如果你还在学习Agent框架的基本概念也可以看但建议先把一个最小可运行的Agent跑起来再来对照这里的机制拆解体会会更深。我下面会按照一条完整的执行链路来讲先讲循环执行是怎么转起来的再讲上下文在这个循环里怎么组织然后是检查点怎么写、任务怎么恢复最后是资源管控。每一部分都会给出我实际用过的方案、踩过的坑以及为什么这样设计。不堆概念只讲能落地的东西。2. 循环执行Agent的心跳是怎么跳的2.1 一次完整的执行循环包含哪些阶段很多人对Agent循环的理解停留在思考-行动-观察这个三段式。这个抽象没错但太粗了。真正落到代码里一个可用的循环至少包含六个阶段状态加载从内存或持久化存储中读取当前任务的完整状态包括历史消息、已完成的步骤、当前进度指针。上下文组装根据当前状态从记忆系统中筛选、裁剪、拼接出这一轮要送给模型的上下文。模型推理调用大模型拿到输出。输出可能是最终答案也可能是一个工具调用请求。动作解析与校验解析模型输出校验工具名是否存在、参数是否合法、是否触发了安全策略。工具执行真正调用外部工具或函数拿到执行结果。这一步是最容易出错的网络超时、权限不足、返回格式异常都可能发生。状态更新与检查点写入把这一轮的结果合并进状态判断任务是否结束如果没结束就写入检查点然后回到第2步。这六个阶段里第2步和第6步是最容易被忽视的。很多人写Agent循环上下文就是简单地把所有历史消息拼起来状态更新就是往列表里append一条。短任务没问题一旦循环超过二十轮上下文爆炸和状态不一致就会同时出现。2.2 循环终止条件的设计不只是任务完成循环什么时候停大部分人的第一反应是模型返回最终答案就停。但实际生产环境里终止条件至少要覆盖四种情况正常完成模型明确返回了终止信号任务目标达成。步数上限设置一个最大循环次数比如50步。防止模型陷入死循环反复调用同一个工具却拿不到有效结果。资源耗尽Token预算用完、外部API调用次数达到配额、执行时间超过阈值。不可恢复错误连续多次工具调用失败或者触发了安全策略必须人工介入。我见过一个典型的坑只设了步数上限没设Token预算。结果Agent在第30步的时候单次上下文已经膨胀到十几万Token一次推理就把预算烧光了后面的步骤全部失败。所以终止条件必须是多维度的任何一个维度触发都应该优雅退出并且把当前状态完整保存下来方便后续恢复。提示终止条件触发时不要直接抛异常退出。应该走一个统一的收尾流程——写入最终检查点、记录终止原因、释放资源、返回一个结构化的终止报告。这个报告在排查问题时非常有用。2.3 循环中的幂等性为什么同一个工具会被调用两次这是我在实际项目里踩过的最隐蔽的坑。Agent在第15步调用了一个创建订单的工具工具执行成功了但在写检查点之前进程崩溃。恢复任务后Agent从第14步的检查点重新开始模型看到历史里没有订单创建成功的记录于是又调用了一次创建订单。结果就是重复下单。解决这个问题的核心是幂等性设计。具体做法有两层第一层给每个工具调用分配一个唯一的调用ID这个ID由任务ID步骤序号工具名参数哈希生成。工具执行前先查一下这个ID是否已经执行过如果执行过就直接返回缓存结果不再真正调用。第二层检查点的写入时机要尽量靠近工具执行完成之后。理想情况下工具执行成功和检查点写入应该在一个原子操作里完成。如果做不到原子操作那至少要在工具执行前先写一条意图记录标记这个调用即将发生恢复时先检查意图记录的状态。# 幂等工具调用的简化实现 def execute_tool_with_idempotency(task_id, step, tool_name, params): call_id generate_call_id(task_id, step, tool_name, params) # 检查是否已经执行过 cached checkpoint_store.get_tool_result(call_id) if cached: return cached # 写入意图记录 checkpoint_store.mark_intent(call_id, statuspending) try: result tool_registry[tool_name](**params) checkpoint_store.save_tool_result(call_id, result) checkpoint_store.mark_intent(call_id, statuscompleted) return result except Exception as e: checkpoint_store.mark_intent(call_id, statusfailed, errorstr(e)) raise这段代码看起来简单但它解决的是Agent从玩具变成工具的关键一步。没有幂等性保障任何涉及副作用的操作都不敢让Agent自动执行。3. 上下文工程在有限窗口里装下整个任务3.1 上下文不是越多越好一个反直觉的结论大模型的上下文窗口越来越大从4K到128K再到某些模型宣称的1M。很多人觉得窗口大了上下文工程就不重要了把所有历史都塞进去就行。但实际测试下来结论恰恰相反上下文越长模型的注意力越容易分散关键信息的召回率反而下降。我做过一个对比实验同一个多步推理任务分别用全量历史和滑动窗口摘要两种上下文策略。全量历史在步骤少的时候表现更好但超过15步之后模型开始遗忘早期的关键约束出现重复提问、忽略已确认信息的情况。而滑动窗口策略虽然丢失了部分细节但通过摘要保留了关键决策点整体成功率反而更高。所以上下文工程的核心不是能装多少而是该装什么。在Agent循环里每一轮送给模型的上下文应该包含四类信息系统指令角色定义、行为约束、输出格式要求。这部分永远放在最前面不参与裁剪。任务目标当前任务的原始描述和验收标准。这部分也不能丢否则模型会跑偏。近期交互最近N轮的模型输出和工具结果。N的取值取决于任务复杂度一般5到10轮。历史摘要对更早轮次的压缩摘要保留关键决策、已确认事实、未解决问题。3.2 上下文裁剪的三种策略与适用场景裁剪策略的选择直接影响到Agent的表现。我常用的有三种滑动窗口只保留最近N轮简单粗暴。适合步骤之间独立性较强的任务比如批量数据处理。缺点是早期的重要约束可能被丢掉。摘要压缩用一个小模型或者规则引擎把早期轮次压缩成一段摘要。适合需要长期记忆的任务比如多轮对话式的问题排查。缺点是摘要本身可能丢失细节而且增加了一次模型调用。关键信息提取不压缩全文而是从历史中提取结构化信息——已完成的步骤列表、已确认的事实、待解决的问题。适合流程明确的任务比如订单处理、工单流转。缺点是需要针对任务类型设计提取规则。实际项目里我通常是组合使用系统指令和任务目标永远保留近期交互用滑动窗口更早的历史用摘要压缩同时维护一个结构化的任务状态面板作为补充。策略优点缺点适用场景滑动窗口实现简单无额外开销早期信息丢失步骤独立性强的任务摘要压缩保留长期记忆增加模型调用可能丢细节多轮对话式任务关键信息提取信息密度高结构清晰需要定制提取规则流程明确的结构化任务3.3 上下文数据流从状态到提示词的组装过程把上面的策略串起来一个完整的上下文组装流程是这样的从状态存储中读取当前任务的完整状态对象。提取系统指令和任务目标作为固定前缀。根据当前步骤序号确定近期交互的窗口范围。如果窗口之前还有历史调用摘要模块生成或读取已有摘要。从记忆系统中检索与当前步骤相关的长期记忆如果有的话。把所有部分按优先级拼接检查总Token数是否超限。如果超限按优先级从低到高裁剪——先裁长期记忆再裁摘要最后才动近期交互。这个流程里有一个容易被忽略的点Token计数要在组装前预估而不是组装后检查。因为组装本身可能触发摘要生成如果组装完才发现超限摘要就白生成了。我的做法是维护一个Token预算表每个部分分配一个上限组装时按预算取用。注意不同模型的Token计算方式不同中文和英文的Token比例也不一样。不要用字符数估算Token一定要用对应模型的Tokenizer实际计算。我见过因为用字符数估算导致实际超出窗口限制、请求被截断的案例。4. 检查点机制让Agent的任务可以存档读档4.1 检查点应该存什么状态快照的完整清单检查点不是简单地把消息列表存下来。一个能支撑任务恢复的检查点至少包含以下内容任务元信息任务ID、创建时间、当前步骤序号、任务状态运行中/已完成/已失败/已暂停。完整消息历史所有轮次的模型输出和工具结果按时间顺序排列。工具调用记录每次工具调用的ID、名称、参数、结果、执行状态。上下文摘要当前使用的摘要内容避免恢复时重新生成。资源消耗统计已用Token数、已用时间、已调用工具次数。自定义状态任务特定的中间变量比如已处理的文件列表、已确认的配置项。这份清单看起来多但每一项在恢复时都可能用到。少存了消息历史恢复后模型不知道之前发生了什么少存了工具调用记录幂等性检查就做不了少存了资源统计恢复后可能直接超预算。4.2 写入时机什么时候写检查点最合适检查点的写入时机是一个权衡。写得太频繁性能开销大写得太少恢复时丢失的进度多。我的经验是三个关键节点必须写每个步骤完成后这是最基本的。不管这一步是模型推理还是工具调用只要状态发生了实质性变化就应该写入检查点。工具调用前对于有副作用的工具执行前先写一条意图记录。这样即使执行过程中崩溃恢复时也能知道这个调用可能已经发生了。循环终止时不管是正常完成还是异常退出都要写一个最终检查点记录终止原因和最终状态。写入方式上我倾向于追加写定期压缩。每次检查点作为一个新记录追加到存储里而不是覆盖旧记录。这样即使某次写入损坏也能回退到上一个有效检查点。定期比如每10个检查点做一次压缩只保留最新的完整快照和之后的增量记录。# 检查点存储的简化结构 class CheckpointStore: def __init__(self, storage_backend): self.backend storage_backend def save(self, task_id, step, state, checkpoint_typeincremental): record { task_id: task_id, step: step, timestamp: time.time(), type: checkpoint_type, state: state } self.backend.append(fcheckpoint:{task_id}, record) def load_latest(self, task_id): records self.backend.read_all(fcheckpoint:{task_id}) # 从后往前找最新的完整快照 for record in reversed(records): if record[type] full: return record[state] return None def load_at_step(self, task_id, step): records self.backend.read_all(fcheckpoint:{task_id}) # 找到目标步骤之前最近的完整快照然后重放增量 base None increments [] for record in records: if record[step] step: if record[type] full: base record[state] increments [] else: increments.append(record) if base is None: return None for inc in increments: base apply_increment(base, inc[state]) return base4.3 存储选型内存、文件还是数据库检查点存哪里取决于任务的生命周期和恢复需求纯内存最快但进程崩溃就没了。只适合短任务或者可以容忍重跑的场景。本地文件实现简单适合单机部署。缺点是并发写入需要加锁多实例部署时无法共享。我一般用JSON Lines格式每行一个检查点记录追加写性能很好。关系型数据库适合需要事务保证的场景。工具执行和检查点写入可以放在同一个事务里保证原子性。缺点是写入延迟比文件高。对象存储适合分布式部署多个Worker可以共享检查点。缺点是读取延迟高不适合高频写入。实际项目里我常用的是本地文件定期同步到对象存储的组合。运行时用本地文件保证低延迟每隔一段时间把检查点同步到对象存储做备份。这样既保证了性能又有了容灾能力。5. 任务恢复从检查点重新拉起一个Agent5.1 恢复流程的完整链路任务恢复不是简单地读取检查点然后继续跑。完整的恢复流程包含以下步骤加载检查点根据任务ID找到最新的有效检查点加载状态。状态校验检查状态的完整性比如消息历史是否连续、工具调用记录是否完整。幂等性检查对于检查点中标记为pending的工具调用查询实际执行结果决定是重试还是跳过。上下文重建根据加载的状态重新组装当前轮次的上下文。资源重置恢复资源计数器但要注意累计消耗不能清零否则会超出总预算。循环重启从检查点记录的步骤序号继续执行循环。这里面最容易出问题的是第3步。如果工具调用没有实现幂等性恢复时就无法判断那个pending的调用到底执行了没有。所以我在前面强调幂等性是任务恢复的前提。5.2 恢复时的状态一致性校验恢复后的状态必须和崩溃前的状态一致否则Agent的行为会变得不可预测。我通常做三类校验消息序列校验检查消息的角色是否交替出现user/assistant/tool有没有缺失的轮次。如果发现消息序列断裂说明检查点不完整需要回退到更早的检查点。工具调用配对校验每个工具调用请求都应该有一个对应的结果。如果发现有请求没结果说明崩溃发生在工具执行过程中需要根据幂等性记录决定如何处理。资源计数校验检查已用Token数、已用时间是否和消息历史匹配。如果差异过大说明检查点可能被篡改或损坏。def validate_checkpoint(state): errors [] # 消息序列校验 messages state[messages] for i in range(1, len(messages)): if messages[i][role] messages[i-1][role]: errors.append(f消息角色重复: 位置{i}) # 工具调用配对校验 pending_calls [c for c in state[tool_calls] if c[status] pending] if pending_calls: errors.append(f存在未完成的工具调用: {len(pending_calls)}个) # 资源计数校验 if state[resources][tokens_used] 0: errors.append(Token计数异常) return errors5.3 恢复失败的降级策略不是所有恢复都能成功。如果检查点损坏、状态不一致、或者幂等性检查无法确定就需要降级策略回退到更早的检查点如果最新检查点有问题尝试加载上一个完整快照。代价是丢失部分进度但至少能继续跑。人工介入对于涉及资金、权限等敏感操作的任务恢复失败时应该暂停并通知人工处理而不是自动重试。重新开始如果任务本身是幂等的而且重跑成本可接受直接从头开始也是一种选择。但要在日志里明确记录避免无限重试。提示恢复失败时一定要把失败原因和当前状态完整记录下来。我见过因为恢复失败后直接丢弃状态导致问题无法复现的情况。状态快照本身就是排查问题的重要线索。6. 资源管控别让Agent烧光你的预算6.1 Token预算的分层管理Token是Agent最核心的资源。我的做法是三层预算管理任务级预算整个任务允许消耗的最大Token数。比如一个任务给50万Token用完就停。步骤级预算单次模型调用允许的最大Token数。防止某一步上下文膨胀导致单次消耗过大。预留预算留出一部分Token用于收尾工作比如生成最终报告、写入检查点。这部分预算不能被正常步骤占用。具体分配上我一般按任务复杂度估算。一个中等复杂度的任务大概20到30步每步平均消耗5000到10000 Token加上收尾预留总预算设在30万到50万比较合理。当然这只是一个起点实际运行后要根据统计数据调整。6.2 工具调用的配额与限流除了Token外部工具的调用次数也是需要管控的资源。特别是那些按次计费的API如果不加限制Agent可能在循环中反复调用产生意外费用。我的做法是给每个工具配置一个配额包括单任务最大调用次数和单位时间最大调用频率。配额检查放在工具执行前的校验阶段超限就直接返回错误让模型知道这个工具暂时不可用引导它换一种方式解决问题。class ToolQuotaManager: def __init__(self, config): self.quotas config # {tool_name: {max_per_task: N, max_per_minute: M}} self.usage {} # 运行时统计 def check_and_consume(self, task_id, tool_name): quota self.quotas.get(tool_name) if not quota: return True # 未配置配额的工具不限制 key f{task_id}:{tool_name} usage self.usage.get(key, {task_count: 0, recent_calls: []}) # 任务级配额检查 if usage[task_count] quota[max_per_task]: return False # 频率检查 now time.time() recent [t for t in usage[recent_calls] if now - t 60] if len(recent) quota[max_per_minute]: return False # 消费配额 usage[task_count] 1 usage[recent_calls] recent [now] self.usage[key] usage return True6.3 超时与熔断防止Agent卡死Agent卡死是另一个常见问题。模型可能陷入无限循环工具可能一直不返回网络可能一直超时。没有超时和熔断机制Agent就会一直挂在那里占用资源却不产生任何进展。我的配置是单次模型调用超时60秒。超过就重试重试两次还失败就终止任务。单次工具调用超时根据工具类型设置一般30秒。超时后标记为失败让模型决定下一步。连续失败熔断如果连续3次工具调用失败暂停任务并写入检查点等待人工介入。总执行时间上限比如30分钟。超过就强制终止保存状态。这些阈值不是拍脑袋定的要根据实际任务的统计分布来调整。我一般会先跑一批测试任务收集每次调用的耗时分布然后取P99作为超时阈值取P95作为熔断参考。管控维度阈值设置触发动作调整依据模型调用超时60秒重试2次后终止P99耗时工具调用超时30秒标记失败继续循环工具类型连续失败熔断3次暂停任务人工介入错误率统计总执行时间30分钟强制终止保存状态任务复杂度7. 把五个环节串起来一个可运行的Agent骨架7.1 核心循环的伪代码实现把前面讲的机制整合起来一个具备上下文管理、检查点、恢复和资源管控的Agent循环大致长这样class AgentRuntime: def __init__(self, config): self.config config self.checkpoint_store CheckpointStore(config.storage) self.context_builder ContextBuilder(config.context) self.resource_manager ResourceManager(config.resources) self.tool_registry ToolRegistry(config.tools) def run(self, task_id, task_goal): # 尝试恢复如果没有检查点则新建 state self.checkpoint_store.load_latest(task_id) if state is None: state self._init_state(task_id, task_goal) while not self._should_terminate(state): # 资源检查 if not self.resource_manager.check(state): state[termination_reason] resource_exhausted break # 组装上下文 context self.context_builder.build(state) # 模型推理 try: output self._call_model(context) except TimeoutError: state[consecutive_failures] 1 if state[consecutive_failures] 3: state[termination_reason] model_timeout break continue # 解析输出 action self._parse_action(output) if action[type] final_answer: state[messages].append(output) state[termination_reason] completed break if action[type] tool_call: # 配额检查 if not self.resource_manager.check_tool_quota(task_id, action[tool]): state[messages].append(self._tool_quota_error(action)) continue # 幂等执行 result self._execute_tool_idempotent(state, action) state[messages].append(result) # 更新状态并写检查点 state[step] 1 self.checkpoint_store.save(task_id, state[step], state) # 收尾 self.checkpoint_store.save(task_id, state[step], state, checkpoint_typefull) return self._build_result(state)这个骨架不完整但把关键机制都串起来了。实际实现时每个模块都需要根据具体场景细化。7.2 各环节的配置参数速查环节关键参数建议值说明循环执行max_steps50根据任务复杂度调整循环执行termination_conditions完成/步数/资源/错误多维度终止上下文recent_window5-10轮近期交互保留轮数上下文summary_threshold10轮超过后触发摘要检查点write_interval每步每步完成后写入检查点full_snapshot_interval每10个检查点定期完整快照任务恢复max_retries3恢复失败重试次数资源管控token_budget30-50万任务级Token预算资源管控tool_timeout30秒单次工具调用超时资源管控circuit_breaker连续3次失败熔断阈值7.3 从Demo到生产还需要补哪些能力上面这套机制能让Agent在单机上稳定运行但要上生产环境还有几件事要做可观测性每一步的输入输出、资源消耗、耗时都要有结构化日志。我一般用JSON格式记录方便后续做分析和告警。并发控制多个任务同时运行时检查点存储、工具配额、模型调用都需要考虑并发安全。数据库事务或者分布式锁是常见方案。版本管理Agent的提示词、工具定义、配置参数都会迭代。检查点里要记录版本号恢复时用对应版本的状态结构避免版本不兼容。安全审计涉及敏感操作的工具调用要有完整的审计日志记录谁在什么时候触发了什么操作。这些能力不是一蹴而就的可以随着业务需求逐步补齐。但循环执行、上下文、检查点、恢复、资源管控这五个基础环节建议在项目早期就设计好不然后期改造的成本会很高。8. 几个我踩过的坑和对应的解法8.1 检查点写入和工具执行之间的竞态早期我做检查点的时候是在工具执行完成后异步写入。结果有一次工具执行成功、检查点还没写完进程就被Kill了。恢复时从上一个检查点开始工具被重复执行。解法就是前面说的工具执行前先写意图记录执行后更新状态。意图记录和工具执行之间虽然还有窗口但至少能通过意图记录判断这个调用可能已经发生了从而在恢复时做人工确认或者幂等重试。8.2 上下文摘要丢失关键约束有一次任务要求所有金额必须保留两位小数这个约束在任务目标里。但跑到第20步的时候摘要模块把任务目标也压缩了模型开始输出整数金额。排查了很久才发现是摘要的问题。解法是任务目标和系统指令永远不参与摘要压缩。摘要只压缩中间的工具调用结果和模型推理过程。这个规则后来成了我的上下文组装的硬性约束。8.3 资源计数在恢复后清零恢复任务时我一开始只恢复了消息历史忘了恢复资源计数器。结果恢复后的任务又从零开始计算Token消耗最终超出了总预算。解法是资源计数器必须作为检查点的一部分持久化。恢复时读取累计值而不是重置。同时恢复本身也要消耗资源这部分也要计入。8.4 工具配额在分布式环境下失效单机运行时工具配额用内存字典就能管住。但部署到多个实例后每个实例各算各的总配额就失控了。解法是配额管理要集中化。可以用Redis的原子操作做计数或者用数据库的行锁。关键是配额的检查和消费必须是原子的不能先检查后消费。9. 写在最后的一点个人体会Agent的运行机制不是一个设计完就完事的东西它更像是一个需要持续调优的系统。我现在的习惯是每上线一个新Agent任务先跑一批测试用例收集循环步数分布、Token消耗分布、工具调用成功率、恢复成功率这些指标然后根据数据调整参数。比如发现平均步数是15步那max_steps设50就够用了发现某个工具的超时率很高就单独给它调大超时阈值。另外不要追求一步到位。我见过有人一开始就想把检查点、恢复、资源管控全部做完美结果项目迟迟上不了线。更务实的做法是先让Agent能跑通基本循环然后加上检查点保证崩溃不丢状态再加资源管控防止预算失控最后补恢复和幂等性。每一步都解决一个具体问题迭代着来。这套机制不只适用于Agent任何长时运行的自动化任务都可以参考。核心思想就一句话把执行过程当成一个可以随时暂停、随时恢复的状态机来设计而不是一个从头跑到尾的脚本。想清楚这一点很多设计决策就自然清晰了。 Agent循环执行实战：上下文、检查点、任务恢复与资源管控
1. 从一次任务中断说起Agent循环执行的真实痛点凌晨两点我盯着日志里那行agent execution terminated due to error发呆。一个跑了四十多分钟的自动化任务在第37步调用外部接口时因为网络抖动直接挂掉前面36步积累的上下文、中间产物、工具调用记录全部归零。第二天重跑又卡在同一个位置。这种场景做过Agent开发的人应该都不陌生——我们花了大量精力在提示词工程、工具编排、模型选型上却常常忽略一个更底层的问题Agent的运行机制本身是否健壮。这篇内容想聊的就是这件事。标题里的几个词——上下文、检查点、任务恢复、循环执行、资源管控——看起来像是五个独立话题实际上它们是一条链上的五个环节。一个能真正跑在生产环境里的Agent不是能调用工具、能返回结果就完事了而是要在循环执行的过程中持续维护上下文在关键节点写入检查点在异常发生后能恢复任务同时全程做好资源管控。缺任何一环Agent都只能停留在Demo阶段。适合谁看如果你正在做Agent开发已经跑通过简单的工具调用链路但一遇到长任务、多轮循环、外部依赖不稳定就出问题那这篇内容就是给你准备的。如果你还在学习Agent框架的基本概念也可以看但建议先把一个最小可运行的Agent跑起来再来对照这里的机制拆解体会会更深。我下面会按照一条完整的执行链路来讲先讲循环执行是怎么转起来的再讲上下文在这个循环里怎么组织然后是检查点怎么写、任务怎么恢复最后是资源管控。每一部分都会给出我实际用过的方案、踩过的坑以及为什么这样设计。不堆概念只讲能落地的东西。2. 循环执行Agent的心跳是怎么跳的2.1 一次完整的执行循环包含哪些阶段很多人对Agent循环的理解停留在思考-行动-观察这个三段式。这个抽象没错但太粗了。真正落到代码里一个可用的循环至少包含六个阶段状态加载从内存或持久化存储中读取当前任务的完整状态包括历史消息、已完成的步骤、当前进度指针。上下文组装根据当前状态从记忆系统中筛选、裁剪、拼接出这一轮要送给模型的上下文。模型推理调用大模型拿到输出。输出可能是最终答案也可能是一个工具调用请求。动作解析与校验解析模型输出校验工具名是否存在、参数是否合法、是否触发了安全策略。工具执行真正调用外部工具或函数拿到执行结果。这一步是最容易出错的网络超时、权限不足、返回格式异常都可能发生。状态更新与检查点写入把这一轮的结果合并进状态判断任务是否结束如果没结束就写入检查点然后回到第2步。这六个阶段里第2步和第6步是最容易被忽视的。很多人写Agent循环上下文就是简单地把所有历史消息拼起来状态更新就是往列表里append一条。短任务没问题一旦循环超过二十轮上下文爆炸和状态不一致就会同时出现。2.2 循环终止条件的设计不只是任务完成循环什么时候停大部分人的第一反应是模型返回最终答案就停。但实际生产环境里终止条件至少要覆盖四种情况正常完成模型明确返回了终止信号任务目标达成。步数上限设置一个最大循环次数比如50步。防止模型陷入死循环反复调用同一个工具却拿不到有效结果。资源耗尽Token预算用完、外部API调用次数达到配额、执行时间超过阈值。不可恢复错误连续多次工具调用失败或者触发了安全策略必须人工介入。我见过一个典型的坑只设了步数上限没设Token预算。结果Agent在第30步的时候单次上下文已经膨胀到十几万Token一次推理就把预算烧光了后面的步骤全部失败。所以终止条件必须是多维度的任何一个维度触发都应该优雅退出并且把当前状态完整保存下来方便后续恢复。提示终止条件触发时不要直接抛异常退出。应该走一个统一的收尾流程——写入最终检查点、记录终止原因、释放资源、返回一个结构化的终止报告。这个报告在排查问题时非常有用。2.3 循环中的幂等性为什么同一个工具会被调用两次这是我在实际项目里踩过的最隐蔽的坑。Agent在第15步调用了一个创建订单的工具工具执行成功了但在写检查点之前进程崩溃。恢复任务后Agent从第14步的检查点重新开始模型看到历史里没有订单创建成功的记录于是又调用了一次创建订单。结果就是重复下单。解决这个问题的核心是幂等性设计。具体做法有两层第一层给每个工具调用分配一个唯一的调用ID这个ID由任务ID步骤序号工具名参数哈希生成。工具执行前先查一下这个ID是否已经执行过如果执行过就直接返回缓存结果不再真正调用。第二层检查点的写入时机要尽量靠近工具执行完成之后。理想情况下工具执行成功和检查点写入应该在一个原子操作里完成。如果做不到原子操作那至少要在工具执行前先写一条意图记录标记这个调用即将发生恢复时先检查意图记录的状态。# 幂等工具调用的简化实现 def execute_tool_with_idempotency(task_id, step, tool_name, params): call_id generate_call_id(task_id, step, tool_name, params) # 检查是否已经执行过 cached checkpoint_store.get_tool_result(call_id) if cached: return cached # 写入意图记录 checkpoint_store.mark_intent(call_id, statuspending) try: result tool_registry[tool_name](**params) checkpoint_store.save_tool_result(call_id, result) checkpoint_store.mark_intent(call_id, statuscompleted) return result except Exception as e: checkpoint_store.mark_intent(call_id, statusfailed, errorstr(e)) raise这段代码看起来简单但它解决的是Agent从玩具变成工具的关键一步。没有幂等性保障任何涉及副作用的操作都不敢让Agent自动执行。3. 上下文工程在有限窗口里装下整个任务3.1 上下文不是越多越好一个反直觉的结论大模型的上下文窗口越来越大从4K到128K再到某些模型宣称的1M。很多人觉得窗口大了上下文工程就不重要了把所有历史都塞进去就行。但实际测试下来结论恰恰相反上下文越长模型的注意力越容易分散关键信息的召回率反而下降。我做过一个对比实验同一个多步推理任务分别用全量历史和滑动窗口摘要两种上下文策略。全量历史在步骤少的时候表现更好但超过15步之后模型开始遗忘早期的关键约束出现重复提问、忽略已确认信息的情况。而滑动窗口策略虽然丢失了部分细节但通过摘要保留了关键决策点整体成功率反而更高。所以上下文工程的核心不是能装多少而是该装什么。在Agent循环里每一轮送给模型的上下文应该包含四类信息系统指令角色定义、行为约束、输出格式要求。这部分永远放在最前面不参与裁剪。任务目标当前任务的原始描述和验收标准。这部分也不能丢否则模型会跑偏。近期交互最近N轮的模型输出和工具结果。N的取值取决于任务复杂度一般5到10轮。历史摘要对更早轮次的压缩摘要保留关键决策、已确认事实、未解决问题。3.2 上下文裁剪的三种策略与适用场景裁剪策略的选择直接影响到Agent的表现。我常用的有三种滑动窗口只保留最近N轮简单粗暴。适合步骤之间独立性较强的任务比如批量数据处理。缺点是早期的重要约束可能被丢掉。摘要压缩用一个小模型或者规则引擎把早期轮次压缩成一段摘要。适合需要长期记忆的任务比如多轮对话式的问题排查。缺点是摘要本身可能丢失细节而且增加了一次模型调用。关键信息提取不压缩全文而是从历史中提取结构化信息——已完成的步骤列表、已确认的事实、待解决的问题。适合流程明确的任务比如订单处理、工单流转。缺点是需要针对任务类型设计提取规则。实际项目里我通常是组合使用系统指令和任务目标永远保留近期交互用滑动窗口更早的历史用摘要压缩同时维护一个结构化的任务状态面板作为补充。策略优点缺点适用场景滑动窗口实现简单无额外开销早期信息丢失步骤独立性强的任务摘要压缩保留长期记忆增加模型调用可能丢细节多轮对话式任务关键信息提取信息密度高结构清晰需要定制提取规则流程明确的结构化任务3.3 上下文数据流从状态到提示词的组装过程把上面的策略串起来一个完整的上下文组装流程是这样的从状态存储中读取当前任务的完整状态对象。提取系统指令和任务目标作为固定前缀。根据当前步骤序号确定近期交互的窗口范围。如果窗口之前还有历史调用摘要模块生成或读取已有摘要。从记忆系统中检索与当前步骤相关的长期记忆如果有的话。把所有部分按优先级拼接检查总Token数是否超限。如果超限按优先级从低到高裁剪——先裁长期记忆再裁摘要最后才动近期交互。这个流程里有一个容易被忽略的点Token计数要在组装前预估而不是组装后检查。因为组装本身可能触发摘要生成如果组装完才发现超限摘要就白生成了。我的做法是维护一个Token预算表每个部分分配一个上限组装时按预算取用。注意不同模型的Token计算方式不同中文和英文的Token比例也不一样。不要用字符数估算Token一定要用对应模型的Tokenizer实际计算。我见过因为用字符数估算导致实际超出窗口限制、请求被截断的案例。4. 检查点机制让Agent的任务可以存档读档4.1 检查点应该存什么状态快照的完整清单检查点不是简单地把消息列表存下来。一个能支撑任务恢复的检查点至少包含以下内容任务元信息任务ID、创建时间、当前步骤序号、任务状态运行中/已完成/已失败/已暂停。完整消息历史所有轮次的模型输出和工具结果按时间顺序排列。工具调用记录每次工具调用的ID、名称、参数、结果、执行状态。上下文摘要当前使用的摘要内容避免恢复时重新生成。资源消耗统计已用Token数、已用时间、已调用工具次数。自定义状态任务特定的中间变量比如已处理的文件列表、已确认的配置项。这份清单看起来多但每一项在恢复时都可能用到。少存了消息历史恢复后模型不知道之前发生了什么少存了工具调用记录幂等性检查就做不了少存了资源统计恢复后可能直接超预算。4.2 写入时机什么时候写检查点最合适检查点的写入时机是一个权衡。写得太频繁性能开销大写得太少恢复时丢失的进度多。我的经验是三个关键节点必须写每个步骤完成后这是最基本的。不管这一步是模型推理还是工具调用只要状态发生了实质性变化就应该写入检查点。工具调用前对于有副作用的工具执行前先写一条意图记录。这样即使执行过程中崩溃恢复时也能知道这个调用可能已经发生了。循环终止时不管是正常完成还是异常退出都要写一个最终检查点记录终止原因和最终状态。写入方式上我倾向于追加写定期压缩。每次检查点作为一个新记录追加到存储里而不是覆盖旧记录。这样即使某次写入损坏也能回退到上一个有效检查点。定期比如每10个检查点做一次压缩只保留最新的完整快照和之后的增量记录。# 检查点存储的简化结构 class CheckpointStore: def __init__(self, storage_backend): self.backend storage_backend def save(self, task_id, step, state, checkpoint_typeincremental): record { task_id: task_id, step: step, timestamp: time.time(), type: checkpoint_type, state: state } self.backend.append(fcheckpoint:{task_id}, record) def load_latest(self, task_id): records self.backend.read_all(fcheckpoint:{task_id}) # 从后往前找最新的完整快照 for record in reversed(records): if record[type] full: return record[state] return None def load_at_step(self, task_id, step): records self.backend.read_all(fcheckpoint:{task_id}) # 找到目标步骤之前最近的完整快照然后重放增量 base None increments [] for record in records: if record[step] step: if record[type] full: base record[state] increments [] else: increments.append(record) if base is None: return None for inc in increments: base apply_increment(base, inc[state]) return base4.3 存储选型内存、文件还是数据库检查点存哪里取决于任务的生命周期和恢复需求纯内存最快但进程崩溃就没了。只适合短任务或者可以容忍重跑的场景。本地文件实现简单适合单机部署。缺点是并发写入需要加锁多实例部署时无法共享。我一般用JSON Lines格式每行一个检查点记录追加写性能很好。关系型数据库适合需要事务保证的场景。工具执行和检查点写入可以放在同一个事务里保证原子性。缺点是写入延迟比文件高。对象存储适合分布式部署多个Worker可以共享检查点。缺点是读取延迟高不适合高频写入。实际项目里我常用的是本地文件定期同步到对象存储的组合。运行时用本地文件保证低延迟每隔一段时间把检查点同步到对象存储做备份。这样既保证了性能又有了容灾能力。5. 任务恢复从检查点重新拉起一个Agent5.1 恢复流程的完整链路任务恢复不是简单地读取检查点然后继续跑。完整的恢复流程包含以下步骤加载检查点根据任务ID找到最新的有效检查点加载状态。状态校验检查状态的完整性比如消息历史是否连续、工具调用记录是否完整。幂等性检查对于检查点中标记为pending的工具调用查询实际执行结果决定是重试还是跳过。上下文重建根据加载的状态重新组装当前轮次的上下文。资源重置恢复资源计数器但要注意累计消耗不能清零否则会超出总预算。循环重启从检查点记录的步骤序号继续执行循环。这里面最容易出问题的是第3步。如果工具调用没有实现幂等性恢复时就无法判断那个pending的调用到底执行了没有。所以我在前面强调幂等性是任务恢复的前提。5.2 恢复时的状态一致性校验恢复后的状态必须和崩溃前的状态一致否则Agent的行为会变得不可预测。我通常做三类校验消息序列校验检查消息的角色是否交替出现user/assistant/tool有没有缺失的轮次。如果发现消息序列断裂说明检查点不完整需要回退到更早的检查点。工具调用配对校验每个工具调用请求都应该有一个对应的结果。如果发现有请求没结果说明崩溃发生在工具执行过程中需要根据幂等性记录决定如何处理。资源计数校验检查已用Token数、已用时间是否和消息历史匹配。如果差异过大说明检查点可能被篡改或损坏。def validate_checkpoint(state): errors [] # 消息序列校验 messages state[messages] for i in range(1, len(messages)): if messages[i][role] messages[i-1][role]: errors.append(f消息角色重复: 位置{i}) # 工具调用配对校验 pending_calls [c for c in state[tool_calls] if c[status] pending] if pending_calls: errors.append(f存在未完成的工具调用: {len(pending_calls)}个) # 资源计数校验 if state[resources][tokens_used] 0: errors.append(Token计数异常) return errors5.3 恢复失败的降级策略不是所有恢复都能成功。如果检查点损坏、状态不一致、或者幂等性检查无法确定就需要降级策略回退到更早的检查点如果最新检查点有问题尝试加载上一个完整快照。代价是丢失部分进度但至少能继续跑。人工介入对于涉及资金、权限等敏感操作的任务恢复失败时应该暂停并通知人工处理而不是自动重试。重新开始如果任务本身是幂等的而且重跑成本可接受直接从头开始也是一种选择。但要在日志里明确记录避免无限重试。提示恢复失败时一定要把失败原因和当前状态完整记录下来。我见过因为恢复失败后直接丢弃状态导致问题无法复现的情况。状态快照本身就是排查问题的重要线索。6. 资源管控别让Agent烧光你的预算6.1 Token预算的分层管理Token是Agent最核心的资源。我的做法是三层预算管理任务级预算整个任务允许消耗的最大Token数。比如一个任务给50万Token用完就停。步骤级预算单次模型调用允许的最大Token数。防止某一步上下文膨胀导致单次消耗过大。预留预算留出一部分Token用于收尾工作比如生成最终报告、写入检查点。这部分预算不能被正常步骤占用。具体分配上我一般按任务复杂度估算。一个中等复杂度的任务大概20到30步每步平均消耗5000到10000 Token加上收尾预留总预算设在30万到50万比较合理。当然这只是一个起点实际运行后要根据统计数据调整。6.2 工具调用的配额与限流除了Token外部工具的调用次数也是需要管控的资源。特别是那些按次计费的API如果不加限制Agent可能在循环中反复调用产生意外费用。我的做法是给每个工具配置一个配额包括单任务最大调用次数和单位时间最大调用频率。配额检查放在工具执行前的校验阶段超限就直接返回错误让模型知道这个工具暂时不可用引导它换一种方式解决问题。class ToolQuotaManager: def __init__(self, config): self.quotas config # {tool_name: {max_per_task: N, max_per_minute: M}} self.usage {} # 运行时统计 def check_and_consume(self, task_id, tool_name): quota self.quotas.get(tool_name) if not quota: return True # 未配置配额的工具不限制 key f{task_id}:{tool_name} usage self.usage.get(key, {task_count: 0, recent_calls: []}) # 任务级配额检查 if usage[task_count] quota[max_per_task]: return False # 频率检查 now time.time() recent [t for t in usage[recent_calls] if now - t 60] if len(recent) quota[max_per_minute]: return False # 消费配额 usage[task_count] 1 usage[recent_calls] recent [now] self.usage[key] usage return True6.3 超时与熔断防止Agent卡死Agent卡死是另一个常见问题。模型可能陷入无限循环工具可能一直不返回网络可能一直超时。没有超时和熔断机制Agent就会一直挂在那里占用资源却不产生任何进展。我的配置是单次模型调用超时60秒。超过就重试重试两次还失败就终止任务。单次工具调用超时根据工具类型设置一般30秒。超时后标记为失败让模型决定下一步。连续失败熔断如果连续3次工具调用失败暂停任务并写入检查点等待人工介入。总执行时间上限比如30分钟。超过就强制终止保存状态。这些阈值不是拍脑袋定的要根据实际任务的统计分布来调整。我一般会先跑一批测试任务收集每次调用的耗时分布然后取P99作为超时阈值取P95作为熔断参考。管控维度阈值设置触发动作调整依据模型调用超时60秒重试2次后终止P99耗时工具调用超时30秒标记失败继续循环工具类型连续失败熔断3次暂停任务人工介入错误率统计总执行时间30分钟强制终止保存状态任务复杂度7. 把五个环节串起来一个可运行的Agent骨架7.1 核心循环的伪代码实现把前面讲的机制整合起来一个具备上下文管理、检查点、恢复和资源管控的Agent循环大致长这样class AgentRuntime: def __init__(self, config): self.config config self.checkpoint_store CheckpointStore(config.storage) self.context_builder ContextBuilder(config.context) self.resource_manager ResourceManager(config.resources) self.tool_registry ToolRegistry(config.tools) def run(self, task_id, task_goal): # 尝试恢复如果没有检查点则新建 state self.checkpoint_store.load_latest(task_id) if state is None: state self._init_state(task_id, task_goal) while not self._should_terminate(state): # 资源检查 if not self.resource_manager.check(state): state[termination_reason] resource_exhausted break # 组装上下文 context self.context_builder.build(state) # 模型推理 try: output self._call_model(context) except TimeoutError: state[consecutive_failures] 1 if state[consecutive_failures] 3: state[termination_reason] model_timeout break continue # 解析输出 action self._parse_action(output) if action[type] final_answer: state[messages].append(output) state[termination_reason] completed break if action[type] tool_call: # 配额检查 if not self.resource_manager.check_tool_quota(task_id, action[tool]): state[messages].append(self._tool_quota_error(action)) continue # 幂等执行 result self._execute_tool_idempotent(state, action) state[messages].append(result) # 更新状态并写检查点 state[step] 1 self.checkpoint_store.save(task_id, state[step], state) # 收尾 self.checkpoint_store.save(task_id, state[step], state, checkpoint_typefull) return self._build_result(state)这个骨架不完整但把关键机制都串起来了。实际实现时每个模块都需要根据具体场景细化。7.2 各环节的配置参数速查环节关键参数建议值说明循环执行max_steps50根据任务复杂度调整循环执行termination_conditions完成/步数/资源/错误多维度终止上下文recent_window5-10轮近期交互保留轮数上下文summary_threshold10轮超过后触发摘要检查点write_interval每步每步完成后写入检查点full_snapshot_interval每10个检查点定期完整快照任务恢复max_retries3恢复失败重试次数资源管控token_budget30-50万任务级Token预算资源管控tool_timeout30秒单次工具调用超时资源管控circuit_breaker连续3次失败熔断阈值7.3 从Demo到生产还需要补哪些能力上面这套机制能让Agent在单机上稳定运行但要上生产环境还有几件事要做可观测性每一步的输入输出、资源消耗、耗时都要有结构化日志。我一般用JSON格式记录方便后续做分析和告警。并发控制多个任务同时运行时检查点存储、工具配额、模型调用都需要考虑并发安全。数据库事务或者分布式锁是常见方案。版本管理Agent的提示词、工具定义、配置参数都会迭代。检查点里要记录版本号恢复时用对应版本的状态结构避免版本不兼容。安全审计涉及敏感操作的工具调用要有完整的审计日志记录谁在什么时候触发了什么操作。这些能力不是一蹴而就的可以随着业务需求逐步补齐。但循环执行、上下文、检查点、恢复、资源管控这五个基础环节建议在项目早期就设计好不然后期改造的成本会很高。8. 几个我踩过的坑和对应的解法8.1 检查点写入和工具执行之间的竞态早期我做检查点的时候是在工具执行完成后异步写入。结果有一次工具执行成功、检查点还没写完进程就被Kill了。恢复时从上一个检查点开始工具被重复执行。解法就是前面说的工具执行前先写意图记录执行后更新状态。意图记录和工具执行之间虽然还有窗口但至少能通过意图记录判断这个调用可能已经发生了从而在恢复时做人工确认或者幂等重试。8.2 上下文摘要丢失关键约束有一次任务要求所有金额必须保留两位小数这个约束在任务目标里。但跑到第20步的时候摘要模块把任务目标也压缩了模型开始输出整数金额。排查了很久才发现是摘要的问题。解法是任务目标和系统指令永远不参与摘要压缩。摘要只压缩中间的工具调用结果和模型推理过程。这个规则后来成了我的上下文组装的硬性约束。8.3 资源计数在恢复后清零恢复任务时我一开始只恢复了消息历史忘了恢复资源计数器。结果恢复后的任务又从零开始计算Token消耗最终超出了总预算。解法是资源计数器必须作为检查点的一部分持久化。恢复时读取累计值而不是重置。同时恢复本身也要消耗资源这部分也要计入。8.4 工具配额在分布式环境下失效单机运行时工具配额用内存字典就能管住。但部署到多个实例后每个实例各算各的总配额就失控了。解法是配额管理要集中化。可以用Redis的原子操作做计数或者用数据库的行锁。关键是配额的检查和消费必须是原子的不能先检查后消费。9. 写在最后的一点个人体会Agent的运行机制不是一个设计完就完事的东西它更像是一个需要持续调优的系统。我现在的习惯是每上线一个新Agent任务先跑一批测试用例收集循环步数分布、Token消耗分布、工具调用成功率、恢复成功率这些指标然后根据数据调整参数。比如发现平均步数是15步那max_steps设50就够用了发现某个工具的超时率很高就单独给它调大超时阈值。另外不要追求一步到位。我见过有人一开始就想把检查点、恢复、资源管控全部做完美结果项目迟迟上不了线。更务实的做法是先让Agent能跑通基本循环然后加上检查点保证崩溃不丢状态再加资源管控防止预算失控最后补恢复和幂等性。每一步都解决一个具体问题迭代着来。这套机制不只适用于Agent任何长时运行的自动化任务都可以参考。核心思想就一句话把执行过程当成一个可以随时暂停、随时恢复的状态机来设计而不是一个从头跑到尾的脚本。想清楚这一点很多设计决策就自然清晰了。