跑过一次长任务型 Agent 的兄弟应该都体会过那种绝望：任务跑到第 7 步，模型突然报上下文超长；重启后一切归零，日志显示最早的几个关键结果已经丢了；更难受的是，这已经是今天第三次在同一个节点崩掉。这不是 Agent 能力不行，而是运行机制没设计好。今天这篇，我想完整地拆一遍 23.3 这套运行机制涉及的五件事：上下文管理、检查点、任务恢复、循环执行和资源管控。它们单独拎出来都不复杂，但串起来就是长任务 Agent 能不能稳定跑完的生死线。这套机制解决的是三类人的痛点：一是做自动化工作流开发的人，二是接大模型做业务系统的团队，三是纯粹想让自己写的 Agent 更靠谱的独立开发者。你们会遇到的问题基本一样：怎么让 Agent 不“失忆”，怎么在崩溃后从断点继续，怎么防止它死循环，怎么控制成本和流量。下面我按实际工程落地顺序，把这套完整流程掰开讲明白。1. 长任务 Agent 最怕什么：整套机制的定位1.1 从一次“崩掉”的 Agent 任务说起我印象很深的是一次数据清洗任务。当时的 Agent 要读取一个几百行的表格，按规则逐条处理，过程中还要调用一个在线接口做分类。跑到第 120 行的时候，接口突然超时，Agent 本身没有自动重试，整个任务直接终止。更尴尬的是，之前的处理结果都存在内存变量里，没有落盘，重启之后全部丢光。后来我统计了一下，那次任务其实已经处理了 96 行，剩下 24 行，大约 15 分钟内就能跑完。但因为缺少检查点，没有任务恢复，这 15 分钟变成了白干。从此之后我所有长任务 Agent 的设计原则就一句话：把每一次运行都当成“可能会中断”来设计。这类问题在纯对话式场景里不明显，因为人类来兜底。但在无人值守的自动化场景里，Agent 需要自己处理中断、超时、上下文增长、资源耗尽，这套运行机制就是为这种情况准备的。1.2 运行机制的五大件：上下文、检查点、任务恢复、循环执行、资源管控这五个东西不是并列的五个模块，而是有严格的依赖关系：上下文管理负责“ Agent 能看到什么”。长任务跑下来，上下文不可能无限制增长，必须设计成有遗忘、有摘要、有外部存储的机制。检查点机制负责“ Agent 当前处于什么状态”。上下文、执行步骤、中间结果都要被固化到持久化存储里。任务恢复负责“中断后如何继续”。恢复 = 加载检查点 + 重建上下文 + 跳过已完成步骤。循环执行负责“ Agent 如何反复行动”。它让 Agent 不是一次性生成一段话就结束，而是像 while 循环一样迭代，直到目标达成。资源管控负责“现有资源够不够继续跑”。Token 预算、API 配额、时间上限、内存占用，都需要在执行过程中动态核算，超了就触发降级或熔断。这五件套配合起来是这样的：每次循环迭代开始前，检查资源余量；迭代结束后，把新产生的上下文增量合并进摘要；然后写检查点；如果中途挂掉，下次启动加载最新的检查点，重建上下文，继续下一轮。整个链路是闭合的。2. 上下文管理：别让 Agent“失忆”也不要“撑爆”2.1 上下文窗口的预算思维大模型 API 有上下文窗口上限，比如常见的 128K、200K 等。但实际可用空间不是窗口上限，而是要扣除系统提示词、工具定义、最近几轮对话、模型输出预留空间。我在 23.3 里直接给上下文做了一个“预算表”：系统提示词：固定占用，通常 2K~4K token工具定义：每个工具的描述和参数 Schema，10 个工具可能就占 3K~5K最近 N 轮对话：按轮次累计，每轮 500~2000 token当前输出预留：至少留 4K~8K token，避免生成到一半截断所以真正的可用历史空间 = 窗口上限 - 系统提示 - 工具定义 - 输出预留。如果窗口是 128K，可能留给历史记录的只有不到 110K。这不是抠门，是为了避免“能输入但生成不了完整答案”的尴尬。我的做法是在每次新增上下文前先估算，估算出超过阈值就直接进入压缩流程。这个阈值不设太满，通常设在窗口上限的 80%。剩下 20% 是给模型输出和临时工具返回的缓冲。2.2 三层上下文架构：原始、摘要、动态裁剪所有长任务 Agent 都会遇到同一个矛盾：历史信息确实有用，但塞不下。23.3 的答案是分层。第一层：原始上下文（最近窗口）。只保留最近几轮完整对话和工具调用结果。这是模型直接能看到的内容，保证它对当前正在做的事情有完整感知。第二层：摘要上下文（历史压缩）。把更早的对话内容总结成一段结构化摘要，比如“已完成 1-10 步骤，结果写入临时存储，关键字段值为 xxx”。摘要要保留决策依据和关键数据，但不能太长。通常控制在 2K~4K token。第三层：持久化存储（外部记忆）。所有中间结果、最终结果、原始记录都存到磁盘、数据库或向量库里。模型不用直接读全部数据，但需要时可以通过检索或工具调用重新加载。这三层中，我最想强调第三层的重要性。很多 Agent 项目只在内存里维护一个 messages 数组，一旦上下文压缩就把数据丢了。正确做法是：任何可能在后继步骤中用到的数据，都必须同步落盘。上下文里只放摘要和引用标识，比如“清洗后的数据存放在 data/result_003.json，共 96 条记录”，而不是把 96 条记录全塞进上下文。2.3 上下文滚动与压缩实践滚动压缩有一个常见的起点：当原始上下文的 token 数超过阈值时，把最早的一半对话提取出来，让模型生成一段摘要，然后把最早的一轮从原始上下文中移除，将摘要插入到上下文开头。这里有个容易踩的坑：摘要一旦生成，原始内容就被移除了，如果摘要生成得不好，后面的步骤会因为信息损失做出错误决策。所以我在摘要生成时强制要求结构化，必须包含“已完成动作”“关键结果”“待办事项”“引用数据路径”四个字段。这样即使细节丢失，Agent 也能通过引用路径去外部存储里取回原始数据。另一个实践是动态裁剪干扰信息。有一些中间调试日志、无关的报错消息，会在上下文里占用大量 token。遇到异常信息时不是直接把完整堆栈塞给模型，而是先做一次截断和提炼，只保留错误码、核心报错信息和当前输入参数。这种裁剪能把错误上下文从 2000 token 压到 300 token，而且对模型诊断更有利。3. 检查点机制：给 Agent“存档”3.1 检查点到底要存什么很多人以为检查点就是把上下文 messages 存下来，这是误解。如果我恢复时只恢复对话内容，那么 Agent 会把它当前内部状态、执行到第几步、哪些工具调用已经发出去过，这些信息全部丢掉，重新开始后它很可能重做之前的工作，甚至重复发邮件。所以我在 23.3 里把检查点设计成一个完整的状态快照，包含几个必须字段：步骤游标（step_index）：当前执行到第几个循环，或者是任务列表中的第几项。恢复后直接从游标继续，不允许重头跑。内部状态（agent_state）：Agent 自己维护的变量、优先级队列、临时标记。原始对话里未必体现，但对决策至关重要。中间结果索引（result_index）：已经产出的数据文件、数据库记录 ID、外部存储路径。已执行动作记录（executed_actions）：每一步对系统的副作用，比如“已经发送邮件给 xxx”“已经写入数据库记录 id=123”。主要用来做幂等去重。上下文摘要（context_summary）：压缩后的历史摘要，配合原始上下文使用。保存完整原始上下文成本太高，保存摘要即可。资源使用计数（resource_meter）：已经消耗的 token 数、API 调用次数、耗时，保证恢复后继续从真实余量扣减。另外别忘了保存当前环境状态。比如外部 API 的认证令牌、临时目录路径、并发锁状态。这些在进程重启后会失效，恢复时需要跳过或重建。3.2 序列化与存储选型序列化格式我调试过多种。JSON 最通用，调试方便，但数据量和性能一般；Pickle 快，但安全和跨版本兼容性差；MessagePack 体积小、速度快，适合生产环境。我最终的方案是：调试阶段用 JSON，生产环境用 MessagePack。如果项目里没有特定要求，直接用 JSON 也完全够用，除非单次检查点已经超过几十 MB。存储位置一般有两个选择：本地文件和数据库。本地文件实现简单，适合单机任务，注意原子写即可；数据库适合分布式和多人协作场景，因为检查点本身是一个很典型的“状态记录”，可以直接用一张 checkpoint 表管理。关键是检查点不能直接在原文件上写，要先写到临时文件，写成功后 rename 覆盖，避免进程中断导致文件写一半变成损坏状态。校验也是不可少的。每次保存检查点时同时写入一个 SHA-256 校验值。加载时先算一遍校验，不一致就回滚到上一个可用检查点。刚开始我觉得多此一举，直到我遇到一次磁盘写入中断，检查点文件直接变成空文件，从那以后校验环节再也没省过。3.3 检查点频率与时机检查点保存得太频繁，比如每轮循环都写一次磁盘，对大模型任务来说其实还好，因为一轮操作可能耗时几十秒到几分钟，写文件的频率不会太高。但如果每轮里面还有多次工具调用，每调用一次就写一个检查点，开销就明显了。我常用的频率策略是三个时机：每完成一个大的任务阶段（比如处理完一批数据，跑完一个子任务）保存一次。每次执行有副作用的工具调用之前（比如发送消息、写库、调用支付接口之前）保存一次，这样失败后恢复不会重复触发副作用。循环边界处保存。在一个完整的 Agent 行动循环结束、准备进入下一轮之前，把状态固化下来。还有一个我自己用的技巧：检查点不只有一个，而是保留最近 N 个。我用一个环形缓冲区，保留最近 5 个检查点。这样即使最新的检查点在恢复时发现已经损坏，或者恢复后逻辑发现状态不对，还能回退到之前第 2、3 个版本。这个 N 可以按任务重要性调整，重要任务保留 10 个。4. 任务恢复：从断点“读档”继续4.1 恢复流程：加载检查点 = 重建上下文 + 恢复内部状态 + 重放游标任务恢复在代码里是一条标准流程，我拆成五步：第一步，读取最新且校验通过的检查点文件。第二步，反序列化，解析出步骤游标、内部状态、结果索引、动作记录、资源计数。第三步，重建上下文。因为检查点里只保存了摘要，恢复时要把摘要作为系统提示的一部分，再额外加载最近一段原始上下文（如果有的话）。如果没有保存原始上下文，就让 Agent 基于摘要继续，同时提示它“你可以通过工具加载历史数据”。第四步，恢复外部状态。比如重新建立数据库连接、刷新 API Token、创建临时目录。第五步，从游标位置继续执行循环。这五步里最容易被忽略的是“提示模型这是恢复运行”而不是“从零开始”。一个务实的技巧：在重建后的上下文中，加一句明确的指令，告诉模型“你之前已经完成了步骤 A、B、C，现在的任务是继续从 D 开始”，这样能有效防止模型觉得自己是新对话，从而提出重复问题或重复执行已完成工作。4.2 幂等性与副作用处理任务恢复最大的坑不在恢复本身，而在恢复之后的“重复执行”。Agent 在崩溃前可能已经调用了一个发通知的工具，然后恢复后又调用一次，结果用户收到两条一模一样的通知。这个问题在检查点机制里必须显式处理。我给每个外部副作用工具都加了两个设计：第一个是请求唯一 ID。每次调用时生成一个 uuid，发送前先记录进“已执行动作记录”，检查点保存后，再真正调用外部工具。恢复时，看到这个 action ID 已经在记录中，就直接跳过，并返回“该操作已执行，结果为 xxx”。第二个是外部系统幂等键。如果调用的是自己可控的接口，就在参数里带一个 idempotency_key，让接口自然去重。如果外部系统不支持，就在本地记录调用参数和返回结果，恢复后如果需要重新调用，必须先对比参数是否完全一致。参数一致但没拿到返回结果的，可以安全重试；已经拿到返回结果的，直接用存储的返回结果，不再调用。这一点对于邮件、短信、支付、写入数据库等操作尤其重要。我在项目里专门把“没有副作用的查询类工具”和“有副作用的写操作工具”分开管理，查询工具天然可重复，写操作必须走幂等检查。4.3 恢复失败怎么办检查点也不是万能的。最常见的失败是检查点文件损坏或者不完整。所以前面说的保留多个检查点非常重要。我在恢复流程里加了一层自动降级：优先用最新检查点；校验失败则用上一个；上一个也失败就再往前找。如果所有检查点都不可用，那就只能从初始状态重新开始，同时保留“这是第二次尝试，请避免重复执行可能造成副作用操作”的提示。另一个恢复失败场景是代码升级导致的检查点不兼容。比如之前保存的内部状态是字典结构，升级后字段改名了，反序列化后拿不到关键值。处理方式是在检查点里加上 schema_version 字段。恢复时检查版本号，如果遇到旧版本，运行一个迁移函数，把旧字段映射到新字段。迁移逻辑最好独立，不要和主流程混在一起。恢复完成后的验证也建议补上：快速做一次“干跑”，比如让 Agent 先列出从当前游标到目标还需要执行哪些步骤，和任务规划比对，确认没有明显缺失，再真正执行。虽然多花一次调用，但能避免带着错误状态继续跑很久。5. 循环执行：让 Agent 在限定范围内反复迭代5.1 循环控制器的设计Agent 的本质是一个“感知-决策-行动”循环。模型每输出一次，只是一个决策点上的反应。要把一个复杂任务做完，必须有一个外部控制器去循环调用模型，把每一次输出作为新的输入，再次交给模型，直到满足终止条件。23.3 里的循环控制器等价于一段 while 循环，循环体是：组装当前上下文调用模型获得决策如果模型要调用工具，执行工具，返回结果更新上下文检查终止条件循环不能没有边界。我在设计时给循环提供了四个终止条件优先级从高到低排列：完成标志：模型显式输出“任务完成”或者成功返回最终结果。最大循环轮次：比如任务最多允许执行 30 轮，超过就强制终止并标记异常。时间预算：任务从开始到当前消耗的总时间超过预算就终止。人工干预信号：控制台或者外部系统发来终止指令。如果触发 2、3、4，那么当前循环的结果不能作为成功结果，需要把 Agent 当前进度保存为检查点，同时生成一条“任务未完成”报告，说明原因、已执行步骤、剩余步骤。这样运营同学可以介入或者调整预算后恢复。5.2 循环中的上下文聚合循环跑多了以后，最大的问题就是每轮输出塞进历史，内容越来越臃肿。我在上下文管理里的思路同样适用于循环：模型每轮的原始输出不全部累积到 messages 里，而是分为“执行日志”和“结果汇总”两类。执行日志只存到磁盘或内存环形缓冲，供审计和排查用。结果汇总是从日志里提炼出来的结构摘要，比如“第 5 轮：调用了数据查询工具，获取到 3 条记录，选定了方案 B”。汇总每轮控制在 100~200 token 内。这种设计能显著延长循环寿命。我见过不加聚合的 Agent，跑到第 10 轮上下文就已经超过 128K；用聚合后，即使跑 30 轮，每轮新增的 token 也就几百个，上下文稳稳保持在预算内。5.3 异常循环与死循环防护死循环是长任务 Agent 最典型的翻车现场。表现是模型反复生成“调用同一个工具”“重复同样的查询”“在同一个问题上打转”。排查下来，很多时候是模型在某个决策点上没有足够信息，它想通过重复查询来获取，但查询结果总是一样的，于是陷入循环。我在循环控制器里设置了“发散检测”逻辑：对最近 N 轮的模型输出做一次相似度比较。如果连续三轮都在调用同一个工具并且传入参数完全一致，就强制中断，改为生成一条诊断消息给上层。诊断消息里要说明“检测到可能的重复循环，下一步建议人工介入”。还有一种情况是模型在工具返回结果后，没有把结果真正纳入决策，而是重新生成一个类似的请求。这种情况一般通过改进上下文聚合来解决，把工具返回的关键结果明确高亮提示给模型。轮次上限是最底层的兜底。但是只给轮次上限不够，因为有些任务本身就需要很多轮次，比如批量处理几百个文件。所以我在资源管控里把时间预算和轮次预算分开，并提供一个“预算用尽但任务可继续”的重试机制：当轮次超限时，保存检查点，然后发送一条“预算不足，是否继续？”的通知，等人确认后把预算调大再恢复。这比直接终止体感好很多。6. 资源管控：别让 Agent 吃光你的算力6.1 资源类型：Token、内存、并发、API 配额、时间资源管控不能只看 Token 消耗，一共有五个维度需要监控：资源典型指标超限后果Token累计消耗、剩余预算直接报错、计费飙升内存进程 RSS、对象占用容器被杀、响应变慢并发同时执行的 Agent 数、工具调用 QPSAPI 限流、连接池耗尽API 配额调用次数、分钟级速率限制429 错误、被服务商封禁时间单轮耗时、总耗时任务延迟、超时中断我的习惯是设计一个 ResourceMonitor 单例，在任务启动时读取初始预算，在每次循环迭代后上报这五类指标的消耗。上报不是简单累加，而是从真实运行环境获取：Token 消耗以模型返回的 usage 为准，内存通过进程信息获取，并发和 QPS 通过计数器获取。6.2 配额分配与动态调整资源配额的初始值来自配置，但执行过程中不能死板地遵守初始值。比如一个任务本来预计消耗 50K Token，跑到一半发现外部数据比预期多，预计还需要 80K。如果死守 50K 就会强制终止，白白浪费前面已经消耗的部分。我在资源管控里增加了一个“动态再规划”机制：当剩余预算低于阈值的 20% 时，暂停执行，运行一个预算法则判断，是调高预算继续，还是进入降级模式。降级模式包括三种策略：摘要模式：把上下文历史压缩得更激进，减少每轮 context 的 token 消耗。探索限制：禁止调用高成本的探索性工具（比如网页搜索、长时间运算），只保留必需的数据查询和写操作。步骤精简：让模型用更少的中间步骤来完成任务，比如跳过不必要的验证环节，直接输出最终结果。动态调整的决策逻辑是“任务价值”优先。如果这是一个重要业务任务，预算翻倍也得上；如果只是数据补全的辅助任务，降级后跑完就算了，不值得追加资源。6.3 全局资源看板与熔断对运行中的 Agent，我还会在全局维护一个资源看板，内容包括：当前所有 Agent 实例的累计 Token 消耗、活跃并发数、平均响应延迟、最近 5 分钟 API 调用次数。看板数据通过一个独立线程定时采样，内存开销很小。熔断机制是资源管控的最后一道防线。当某个指标超过硬上限，比如单实例内存达到容器上限的 90%，或者 API 调用错误率超过 30%，触发熔断。熔断不是直接杀掉进程，而是“暂停-保存检查点-上报”。先把 Agent 冻在当前状态，保存检查点，然后向上层发出资源告警，等待人工决定是扩容后恢复还是中止任务。这种策略在实际运行里救了我很多次。有一次 Python 脚本不小心在循环里累积了巨大的中间对象，内存涨到快溢出，熔断机制把进程冻结并保存了检查点，重启后从检查点恢复，中间结果已经释放，顺利跑完。7. 一套完整流程串起来：23.3 版落地示例7.1 配置示例下面是一份典型的任务配置。我把这个文件命名为 task_config.yaml，放在任务目录下。它的作用不是让开发者手动配置所有细节，而是允许同一个运行器复用不同的任务模板。task: name: data_clean_and_cluster description: "读取 csv 文件，逐行清洗，最后做聚类分析" max_steps: 50 max_duration_sec: 1800 # 30 分钟硬限制 agent: model: "your-model-name" temperature: 0.2 system_prompt_file: "prompts/system.md" context: window_limit: 128000 # 模型窗口上限 reserved_for_output: 8000 # 输出预留 compress_threshold: 0.8 # 达到 80% 就压缩 max_recent_rounds: 8 # 保存最近 8 轮原始对话 summary_max_tokens: 4000 checkpoint: dir: "./checkpoints" format: "json" # json / msgpack keep_last: 5 save_on_stage: true save_before_side_effect: true resource: token_budget: 200000 api_call_budget: 100 max_concurrency: 2 memory_limit_mb: 2048 qps_limit: 5 auto_downgrade: true recovery: max_restarts: 3 verify_on_restart: true其中compress_threshold设成 0.8 表示当上下文估算 token 达到(window_limit - reserved_for_output) * 0.8时，启动压缩。max_recent_rounds控制保留最近 8 轮完整历史。资源预算里，auto_downgrade: true表示动态调整关闭时触发降级策略。7.2 核心代码骨架（Python）下面是一个精简但功能完整的运行器骨架。它把检查点加载、循环控制、资源监控、保存检查点这四件事串在一起。代码不能直接跑，但结构和关键逻辑可以照着改。import hashlib import json import os import time import uuid from dataclasses import dataclass, asdict @dataclass class CheckpointData: schema_version: int = 1 step_index: int = 0 agent_state: dict = None result_index: dict = None executed_actions: list = None context_summary: str = "" resource_meter: dict = None updated_at: float = 0.0 class AgentRuntime: def __init__(self, config, resource_monitor): self.config = config self.monitor = resource_monitor self.ckpt_path = config["checkpoint"]["dir"] os.makedirs(self.ckpt_path, exist_ok=True) def build_context(self, recent_messages, summary, new_step_info): ctx = [] if summary: ctx.append({"role": "system", "content": f"[历史摘要]\n{summary}"}) ctx.extend(recent_messages) ctx.append({"role": "user", "content": f"[当前步骤]\n{new_step_info}"}) return ctx def save_checkpoint(self, step_index, state, result_index, executed_actions, summary, meter, label="step"): data = CheckpointData( step_index=step_index, agent_state=state, result_index=result_index, executed_actions=executed_actions, context_summary=summary, resource_meter=meter, updated_at=time.time(), ) raw = json.dumps(asdict(data), ensure_ascii=False, indent=2).encode("utf-8") checksum = hashlib.sha256(raw).hexdigest() # 先写临时文件再 rename，避免写一半损坏 tmp_path = os.path.join(self.ckpt_path, f"ckpt_{label}.tmp") final_path = os.path.join(self.ckpt_path, f"ckpt_{step_index}_{label}.json") with open(tmp_path, "wb") as f: f.write(raw) with open(tmp_path, "a") as f: f.write(f"\n# checksum: {checksum}") os.rename(tmp_path, final_path) # 清理旧检查点，保留 keep_last 个 self._prune_old_checkpoints(final_path) return final_path def load_latest_checkpoint(self): files = sorted( [f for f in os.listdir(self.ckpt_path) if f.endswith(".json")], key=lambda f: os.path.getmtime(os.path.join(self.ckpt_path, f)), reverse=True, ) for fname in files: path = os.path.join(self.ckpt_path, fname) try: raw = open(path, "r", encoding="utf-8").read() body = raw.split("\n# checksum: ")[0] expect = raw.split("\n# checksum: ")[1].strip() actual = hashlib.sha256(body.encode("utf-8")).hexdigest() if actual != expect: print(f"[recovery] checksum mismatch: {fname}") continue ckpt = json.loads(body) print(f"[recovery] load success: {fname}") return ckpt except Exception as e: print(f"[recovery] load failed: {fname}: {e}") return None def run(self): ckpt = self.load_latest_checkpoint() if ckpt is not None: step_index = ckpt["step_index"] state = ckpt["agent_state"] result_index = ckpt["result_index"] or {} executed = ckpt["executed_actions"] or [] summary = ckpt["context_summary"] or "" meter = ckpt["resource_meter"] print(f"[run] resuming from step {step_index}") else: step_index = 0 state = {"initialized": True} result_index = {} executed = [] summary = "" meter = {"token_used": 0, "calls": 0} print("[run] fresh start") while step_index < self.config["task"]["max_steps"]: if self.monitor.is_budget_exhausted(): print("[run] resource budget exhausted, pausing...") self.save_checkpoint(step_index, state, result_index, executed, summary, meter, label="pause") break if self.monitor.overall_time_exceeded(): print("[run] time budget exceeded, pausing...") break # 组装上下文 recent_messages = self._load_recent_messages(state) step_info = f"第 {step_index} 步，当前任务目标..." ctx = self.build_context(recent_messages, summary, step_info) # 请模型决策（实际调用封装在 model_client 里） decision = self.model_client.decide(ctx) if decision.action == "tool_call": # 副作用工具调用前要检查幂等 action_id = uuid.uuid4().hex if decision.tool_name in result_index and \ result_index[decision.tool_name] == decision.tool_args: print(f"[run] duplicated call skipped: {decision.tool_name}") else: result = self.tool_executor.execute( decision.tool_name, decision.tool_args ) # 记录执行结果 result_index[f"result_{step_index}"] = { "tool": decision.tool_name, "args": decision.tool_args, "result_ref": self._store_result(result), } executed.append(action_id) # 更新上下文摘要 summary = self._update_summary(summary, decision, result) elif decision.action == "finish": print(f"[run] task finished at step {step_index}") self.save_checkpoint(step_index, state, result_index, executed, summary, meter, label="finish") return result_index step_index += 1 meter["calls"] += 1 meter["token_used"] += self.model_client.last_usage() # 循环边界保存检查点 if self.config["checkpoint"]["save_on_stage"]: self.save_checkpoint(step_index, state, result_index, executed, summary, meter, label="stage") print("[run] loop ended without finish")这段骨架我把执行过程简化了不少，但保留了对本文所有关键点的对应关系。你写自己的运行器时，重点也不在代码本身，而在里面那几个条件块：恢复前加载检查点、副作用调用前的幂等判断、预算耗尽时保存暂停检查点。这三条决定了你的 Agent 是不是真的“可恢复”。7.3 一次实测运行记录下面是一次用 23.3 跑真实任务的日志节选，我隐去了具体业务信息：[recovery] load success: ckpt_36_stage.json [run] resuming from step 36 [run] step 36: tool_call -> db_query, cost 1.2s, result 8 rows [run] step 37: tool_call -> external_classify, cost 3.0s, result ok [run] step 38: token_used=152400, remaining=47600, ratio=0.76 [context] compress triggered, summary token 3800 [run] step 39: tool_call -> dedup_merge, cost 0.8s, result 96 rows -> 82 rows [run] step 40: tool_call -> side_effect: send_notification, action_id pre-check skipped [run] step 41: finish, final result stored at result_index可以看到第 36 步是恢复点，说明之前中断过一次，恢复模块正常读档。第 38 步触发了上下文压缩，因为 token 占比到了 76%，压成摘要后继续跑。第 40 步调用了一个有副作用的通知接口，因为检查点里已经有该 action 的记录，所以直接跳过没有重复发送。这里我要说明，第 40 步的“跳过”说明副作用的幂等设计在工作，但也提醒我们：如果崩溃前通知已发出，但用户没收到（比如投递失败），那么跳过会导致漏发。所以对于关键副作用，最好在检查点里同时保存“响应状态”，只有状态为 success 时才跳过，状态为 unknown 的，应重新执行并带上幂等键。8. 常见问题与排查技巧实录我整理了几个高频问题，都是从实际运行里踩过的坑，格式做成速查表，方便你直接对照。现象可能原因排查思路解决与避坑恢复后 Agent 从头开始执行检查点没加载成功或恢复流程没提示模型看日志确认 load success；检查模型是否收到“从第 N 步继续”的指令在重建上下文中显式加入恢复指令，并把步骤游标写入系统提示恢复后重复发送邮件/写入数据副作用没有被幂等记录检查 executed_actions 是否完整；外部系统是否支持幂等键工具调用前先记录 action id；调用后把 state 写入检查点再真正执行上下文压缩后关键信息丢失摘要只记录了过程没记录结果检查摘要中是否包含“结果引用路径”摘要必须强制包含 result_index 中文件的存储路径循环卡在同一个工具调用模型没有从返回结果中获得有效信息对比最近 3 轮的工具参数是否一致开启发散检测，连续相同参数触发中断；把工具结果加工成模型更容易理解的结构检查点文件损坏直接覆盖写文件导致写了一半检查文件大小和校验和使用临时文件+rename；保留多个检查点运行到一半预算耗尽但任务还有价值预算初始值设定过死查看资源看板的剩余 token 和调用次数开启动态再规划，预算充足时扩容，不足时降级内存持续增长循环内中间数据没有释放观察进程 RSS 曲线；检查是否把大对象塞入上下文中间结果落盘，内存只保留引用；设置内存熔断恢复后上下文里全是旧历史，新信息加不进来摘要和原始上下文拼接顺序不对查看 build_context 拼接逻辑把摘要放在最前面，最近对话放在其后，当前步骤放在末尾除了表格里的问题，再分享两个排查习惯。第一个是给每个 Agent 任务打上 trace_id。检查点文件名、日志行、工具调用记录、资源看板全带同一个 trace_id。排查一次“队列里三个任务同时挂掉”的问题时，就是靠 trace_id 定位到某个共享连接池被耗尽导致的连锁失败。如果没有 trace_id，三个任务的日志混在一起，很难定位。第二个是恢复后做“检查点演练”。在绝不影响线上数据的前提下，故意把任务中断一次，然后立即恢复，检查恢复后状态是否正确。这个演练最好在开发环境跑通，否则生产环境第一次恢复时，你就只能看运气了。9. 个人经验与后续扩展我在实际应用这套机制时，最大的体会是把 Agent 当成一个有状态的服务来设计，而不是当成一个无状态的 API 调用。它和游戏存档很像：存档不需要记录你按下的每一次按键，只需要记录你当前在哪张地图、装备了什么、接了什么任务。检查点就是这套机制的“存档”，上下文就是你的“任务日志”，资源管控就是“货币和体力”。想清楚这个类比之后，整台运行机构的每个模块都变得很自然。另外我建议拿到这套代码后，先做一次“最小实现”：一个循环、一个检查点、一个恢复、一个资源计数器，跑一个只有 5 步的小任务，故意在第 3 步杀掉进程再恢复。把这套闭环跑通，再往里面加复杂的调度、多 Agent 协作、动态决策，都会顺很多。23.3 这套机制以后还可以扩展的方向，我目前在想的是把“检查点”本身变成一种可以搜索的状态库。也就是说，不只是最近的检查点，而是让某个 Agent 可以读取其他历史任务的检查点，复用它们的状态摘要和结果索引，减少从零开始规划的成本。另一个方向是针对多 Agent 协作场景做分布式事务，两个 Agent 共享一个任务状态时，检查点不能是单机文件，而要走分布式锁和提交日志。这个我还没完全跑通，等有结论了再来补充。总之，先把单 Agent 的这套机制吃透，比什么都重要。
跑过一次长任务型 Agent 的兄弟应该都体会过那种绝望：任务跑到第 7 步，模型突然报上下文超长；重启后一切归零，日志显示最早的几个关键结果已经丢了；更难受的是，这已经是今天第三次在同一个节点崩掉。这不是 Agent 能力不行，而是运行机制没设计好。今天这篇，我想完整地拆一遍 23.3 这套运行机制涉及的五件事：上下文管理、检查点、任务恢复、循环执行和资源管控。它们单独拎出来都不复杂，但串起来就是长任务 Agent 能不能稳定跑完的生死线。这套机制解决的是三类人的痛点：一是做自动化工作流开发的人，二是接大模型做业务系统的团队，三是纯粹想让自己写的 Agent 更靠谱的独立开发者。你们会遇到的问题基本一样：怎么让 Agent 不“失忆”，怎么在崩溃后从断点继续，怎么防止它死循环，怎么控制成本和流量。下面我按实际工程落地顺序，把这套完整流程掰开讲明白。1. 长任务 Agent 最怕什么：整套机制的定位1.1 从一次“崩掉”的 Agent 任务说起我印象很深的是一次数据清洗任务。当时的 Agent 要读取一个几百行的表格，按规则逐条处理，过程中还要调用一个在线接口做分类。跑到第 120 行的时候，接口突然超时，Agent 本身没有自动重试，整个任务直接终止。更尴尬的是，之前的处理结果都存在内存变量里，没有落盘，重启之后全部丢光。后来我统计了一下，那次任务其实已经处理了 96 行，剩下 24 行，大约 15 分钟内就能跑完。但因为缺少检查点，没有任务恢复，这 15 分钟变成了白干。从此之后我所有长任务 Agent 的设计原则就一句话：把每一次运行都当成“可能会中断”来设计。这类问题在纯对话式场景里不明显，因为人类来兜底。但在无人值守的自动化场景里，Agent 需要自己处理中断、超时、上下文增长、资源耗尽，这套运行机制就是为这种情况准备的。1.2 运行机制的五大件：上下文、检查点、任务恢复、循环执行、资源管控这五个东西不是并列的五个模块，而是有严格的依赖关系：上下文管理负责“ Agent 能看到什么”。长任务跑下来，上下文不可能无限制增长，必须设计成有遗忘、有摘要、有外部存储的机制。检查点机制负责“ Agent 当前处于什么状态”。上下文、执行步骤、中间结果都要被固化到持久化存储里。任务恢复负责“中断后如何继续”。恢复 = 加载检查点 + 重建上下文 + 跳过已完成步骤。循环执行负责“ Agent 如何反复行动”。它让 Agent 不是一次性生成一段话就结束，而是像 while 循环一样迭代，直到目标达成。资源管控负责“现有资源够不够继续跑”。Token 预算、API 配额、时间上限、内存占用，都需要在执行过程中动态核算，超了就触发降级或熔断。这五件套配合起来是这样的：每次循环迭代开始前，检查资源余量；迭代结束后，把新产生的上下文增量合并进摘要；然后写检查点；如果中途挂掉，下次启动加载最新的检查点，重建上下文，继续下一轮。整个链路是闭合的。2. 上下文管理：别让 Agent“失忆”也不要“撑爆”2.1 上下文窗口的预算思维大模型 API 有上下文窗口上限，比如常见的 128K、200K 等。但实际可用空间不是窗口上限，而是要扣除系统提示词、工具定义、最近几轮对话、模型输出预留空间。我在 23.3 里直接给上下文做了一个“预算表”：系统提示词：固定占用，通常 2K~4K token工具定义：每个工具的描述和参数 Schema，10 个工具可能就占 3K~5K最近 N 轮对话：按轮次累计，每轮 500~2000 token当前输出预留：至少留 4K~8K token，避免生成到一半截断所以真正的可用历史空间 = 窗口上限 - 系统提示 - 工具定义 - 输出预留。如果窗口是 128K，可能留给历史记录的只有不到 110K。这不是抠门，是为了避免“能输入但生成不了完整答案”的尴尬。我的做法是在每次新增上下文前先估算，估算出超过阈值就直接进入压缩流程。这个阈值不设太满，通常设在窗口上限的 80%。剩下 20% 是给模型输出和临时工具返回的缓冲。2.2 三层上下文架构：原始、摘要、动态裁剪所有长任务 Agent 都会遇到同一个矛盾：历史信息确实有用，但塞不下。23.3 的答案是分层。第一层：原始上下文（最近窗口）。只保留最近几轮完整对话和工具调用结果。这是模型直接能看到的内容，保证它对当前正在做的事情有完整感知。第二层：摘要上下文（历史压缩）。把更早的对话内容总结成一段结构化摘要，比如“已完成 1-10 步骤，结果写入临时存储，关键字段值为 xxx”。摘要要保留决策依据和关键数据，但不能太长。通常控制在 2K~4K token。第三层：持久化存储（外部记忆）。所有中间结果、最终结果、原始记录都存到磁盘、数据库或向量库里。模型不用直接读全部数据，但需要时可以通过检索或工具调用重新加载。这三层中，我最想强调第三层的重要性。很多 Agent 项目只在内存里维护一个 messages 数组，一旦上下文压缩就把数据丢了。正确做法是：任何可能在后继步骤中用到的数据，都必须同步落盘。上下文里只放摘要和引用标识，比如“清洗后的数据存放在 data/result_003.json，共 96 条记录”，而不是把 96 条记录全塞进上下文。2.3 上下文滚动与压缩实践滚动压缩有一个常见的起点：当原始上下文的 token 数超过阈值时，把最早的一半对话提取出来，让模型生成一段摘要，然后把最早的一轮从原始上下文中移除，将摘要插入到上下文开头。这里有个容易踩的坑：摘要一旦生成，原始内容就被移除了，如果摘要生成得不好，后面的步骤会因为信息损失做出错误决策。所以我在摘要生成时强制要求结构化，必须包含“已完成动作”“关键结果”“待办事项”“引用数据路径”四个字段。这样即使细节丢失，Agent 也能通过引用路径去外部存储里取回原始数据。另一个实践是动态裁剪干扰信息。有一些中间调试日志、无关的报错消息，会在上下文里占用大量 token。遇到异常信息时不是直接把完整堆栈塞给模型，而是先做一次截断和提炼，只保留错误码、核心报错信息和当前输入参数。这种裁剪能把错误上下文从 2000 token 压到 300 token，而且对模型诊断更有利。3. 检查点机制：给 Agent“存档”3.1 检查点到底要存什么很多人以为检查点就是把上下文 messages 存下来，这是误解。如果我恢复时只恢复对话内容，那么 Agent 会把它当前内部状态、执行到第几步、哪些工具调用已经发出去过，这些信息全部丢掉，重新开始后它很可能重做之前的工作，甚至重复发邮件。所以我在 23.3 里把检查点设计成一个完整的状态快照，包含几个必须字段：步骤游标（step_index）：当前执行到第几个循环，或者是任务列表中的第几项。恢复后直接从游标继续，不允许重头跑。内部状态（agent_state）：Agent 自己维护的变量、优先级队列、临时标记。原始对话里未必体现，但对决策至关重要。中间结果索引（result_index）：已经产出的数据文件、数据库记录 ID、外部存储路径。已执行动作记录（executed_actions）：每一步对系统的副作用，比如“已经发送邮件给 xxx”“已经写入数据库记录 id=123”。主要用来做幂等去重。上下文摘要（context_summary）：压缩后的历史摘要，配合原始上下文使用。保存完整原始上下文成本太高，保存摘要即可。资源使用计数（resource_meter）：已经消耗的 token 数、API 调用次数、耗时，保证恢复后继续从真实余量扣减。另外别忘了保存当前环境状态。比如外部 API 的认证令牌、临时目录路径、并发锁状态。这些在进程重启后会失效，恢复时需要跳过或重建。3.2 序列化与存储选型序列化格式我调试过多种。JSON 最通用，调试方便，但数据量和性能一般；Pickle 快，但安全和跨版本兼容性差；MessagePack 体积小、速度快，适合生产环境。我最终的方案是：调试阶段用 JSON，生产环境用 MessagePack。如果项目里没有特定要求，直接用 JSON 也完全够用，除非单次检查点已经超过几十 MB。存储位置一般有两个选择：本地文件和数据库。本地文件实现简单，适合单机任务，注意原子写即可；数据库适合分布式和多人协作场景，因为检查点本身是一个很典型的“状态记录”，可以直接用一张 checkpoint 表管理。关键是检查点不能直接在原文件上写，要先写到临时文件，写成功后 rename 覆盖，避免进程中断导致文件写一半变成损坏状态。校验也是不可少的。每次保存检查点时同时写入一个 SHA-256 校验值。加载时先算一遍校验，不一致就回滚到上一个可用检查点。刚开始我觉得多此一举，直到我遇到一次磁盘写入中断，检查点文件直接变成空文件，从那以后校验环节再也没省过。3.3 检查点频率与时机检查点保存得太频繁，比如每轮循环都写一次磁盘，对大模型任务来说其实还好，因为一轮操作可能耗时几十秒到几分钟，写文件的频率不会太高。但如果每轮里面还有多次工具调用，每调用一次就写一个检查点，开销就明显了。我常用的频率策略是三个时机：每完成一个大的任务阶段（比如处理完一批数据，跑完一个子任务）保存一次。每次执行有副作用的工具调用之前（比如发送消息、写库、调用支付接口之前）保存一次，这样失败后恢复不会重复触发副作用。循环边界处保存。在一个完整的 Agent 行动循环结束、准备进入下一轮之前，把状态固化下来。还有一个我自己用的技巧：检查点不只有一个，而是保留最近 N 个。我用一个环形缓冲区，保留最近 5 个检查点。这样即使最新的检查点在恢复时发现已经损坏，或者恢复后逻辑发现状态不对，还能回退到之前第 2、3 个版本。这个 N 可以按任务重要性调整，重要任务保留 10 个。4. 任务恢复：从断点“读档”继续4.1 恢复流程：加载检查点 = 重建上下文 + 恢复内部状态 + 重放游标任务恢复在代码里是一条标准流程，我拆成五步：第一步，读取最新且校验通过的检查点文件。第二步，反序列化，解析出步骤游标、内部状态、结果索引、动作记录、资源计数。第三步，重建上下文。因为检查点里只保存了摘要，恢复时要把摘要作为系统提示的一部分，再额外加载最近一段原始上下文（如果有的话）。如果没有保存原始上下文，就让 Agent 基于摘要继续，同时提示它“你可以通过工具加载历史数据”。第四步，恢复外部状态。比如重新建立数据库连接、刷新 API Token、创建临时目录。第五步，从游标位置继续执行循环。这五步里最容易被忽略的是“提示模型这是恢复运行”而不是“从零开始”。一个务实的技巧：在重建后的上下文中，加一句明确的指令，告诉模型“你之前已经完成了步骤 A、B、C，现在的任务是继续从 D 开始”，这样能有效防止模型觉得自己是新对话，从而提出重复问题或重复执行已完成工作。4.2 幂等性与副作用处理任务恢复最大的坑不在恢复本身，而在恢复之后的“重复执行”。Agent 在崩溃前可能已经调用了一个发通知的工具，然后恢复后又调用一次，结果用户收到两条一模一样的通知。这个问题在检查点机制里必须显式处理。我给每个外部副作用工具都加了两个设计：第一个是请求唯一 ID。每次调用时生成一个 uuid，发送前先记录进“已执行动作记录”，检查点保存后，再真正调用外部工具。恢复时，看到这个 action ID 已经在记录中，就直接跳过，并返回“该操作已执行，结果为 xxx”。第二个是外部系统幂等键。如果调用的是自己可控的接口，就在参数里带一个 idempotency_key，让接口自然去重。如果外部系统不支持，就在本地记录调用参数和返回结果，恢复后如果需要重新调用，必须先对比参数是否完全一致。参数一致但没拿到返回结果的，可以安全重试；已经拿到返回结果的，直接用存储的返回结果，不再调用。这一点对于邮件、短信、支付、写入数据库等操作尤其重要。我在项目里专门把“没有副作用的查询类工具”和“有副作用的写操作工具”分开管理，查询工具天然可重复，写操作必须走幂等检查。4.3 恢复失败怎么办检查点也不是万能的。最常见的失败是检查点文件损坏或者不完整。所以前面说的保留多个检查点非常重要。我在恢复流程里加了一层自动降级：优先用最新检查点；校验失败则用上一个；上一个也失败就再往前找。如果所有检查点都不可用，那就只能从初始状态重新开始，同时保留“这是第二次尝试，请避免重复执行可能造成副作用操作”的提示。另一个恢复失败场景是代码升级导致的检查点不兼容。比如之前保存的内部状态是字典结构，升级后字段改名了，反序列化后拿不到关键值。处理方式是在检查点里加上 schema_version 字段。恢复时检查版本号，如果遇到旧版本，运行一个迁移函数，把旧字段映射到新字段。迁移逻辑最好独立，不要和主流程混在一起。恢复完成后的验证也建议补上：快速做一次“干跑”，比如让 Agent 先列出从当前游标到目标还需要执行哪些步骤，和任务规划比对，确认没有明显缺失，再真正执行。虽然多花一次调用，但能避免带着错误状态继续跑很久。5. 循环执行：让 Agent 在限定范围内反复迭代5.1 循环控制器的设计Agent 的本质是一个“感知-决策-行动”循环。模型每输出一次，只是一个决策点上的反应。要把一个复杂任务做完，必须有一个外部控制器去循环调用模型，把每一次输出作为新的输入，再次交给模型，直到满足终止条件。23.3 里的循环控制器等价于一段 while 循环，循环体是：组装当前上下文调用模型获得决策如果模型要调用工具，执行工具，返回结果更新上下文检查终止条件循环不能没有边界。我在设计时给循环提供了四个终止条件优先级从高到低排列：完成标志：模型显式输出“任务完成”或者成功返回最终结果。最大循环轮次：比如任务最多允许执行 30 轮，超过就强制终止并标记异常。时间预算：任务从开始到当前消耗的总时间超过预算就终止。人工干预信号：控制台或者外部系统发来终止指令。如果触发 2、3、4，那么当前循环的结果不能作为成功结果，需要把 Agent 当前进度保存为检查点，同时生成一条“任务未完成”报告，说明原因、已执行步骤、剩余步骤。这样运营同学可以介入或者调整预算后恢复。5.2 循环中的上下文聚合循环跑多了以后，最大的问题就是每轮输出塞进历史，内容越来越臃肿。我在上下文管理里的思路同样适用于循环：模型每轮的原始输出不全部累积到 messages 里，而是分为“执行日志”和“结果汇总”两类。执行日志只存到磁盘或内存环形缓冲，供审计和排查用。结果汇总是从日志里提炼出来的结构摘要，比如“第 5 轮：调用了数据查询工具，获取到 3 条记录，选定了方案 B”。汇总每轮控制在 100~200 token 内。这种设计能显著延长循环寿命。我见过不加聚合的 Agent，跑到第 10 轮上下文就已经超过 128K；用聚合后，即使跑 30 轮，每轮新增的 token 也就几百个，上下文稳稳保持在预算内。5.3 异常循环与死循环防护死循环是长任务 Agent 最典型的翻车现场。表现是模型反复生成“调用同一个工具”“重复同样的查询”“在同一个问题上打转”。排查下来，很多时候是模型在某个决策点上没有足够信息，它想通过重复查询来获取，但查询结果总是一样的，于是陷入循环。我在循环控制器里设置了“发散检测”逻辑：对最近 N 轮的模型输出做一次相似度比较。如果连续三轮都在调用同一个工具并且传入参数完全一致，就强制中断，改为生成一条诊断消息给上层。诊断消息里要说明“检测到可能的重复循环，下一步建议人工介入”。还有一种情况是模型在工具返回结果后，没有把结果真正纳入决策，而是重新生成一个类似的请求。这种情况一般通过改进上下文聚合来解决，把工具返回的关键结果明确高亮提示给模型。轮次上限是最底层的兜底。但是只给轮次上限不够，因为有些任务本身就需要很多轮次，比如批量处理几百个文件。所以我在资源管控里把时间预算和轮次预算分开，并提供一个“预算用尽但任务可继续”的重试机制：当轮次超限时，保存检查点，然后发送一条“预算不足，是否继续？”的通知，等人确认后把预算调大再恢复。这比直接终止体感好很多。6. 资源管控：别让 Agent 吃光你的算力6.1 资源类型：Token、内存、并发、API 配额、时间资源管控不能只看 Token 消耗，一共有五个维度需要监控：资源典型指标超限后果Token累计消耗、剩余预算直接报错、计费飙升内存进程 RSS、对象占用容器被杀、响应变慢并发同时执行的 Agent 数、工具调用 QPSAPI 限流、连接池耗尽API 配额调用次数、分钟级速率限制429 错误、被服务商封禁时间单轮耗时、总耗时任务延迟、超时中断我的习惯是设计一个 ResourceMonitor 单例，在任务启动时读取初始预算，在每次循环迭代后上报这五类指标的消耗。上报不是简单累加，而是从真实运行环境获取：Token 消耗以模型返回的 usage 为准，内存通过进程信息获取，并发和 QPS 通过计数器获取。6.2 配额分配与动态调整资源配额的初始值来自配置，但执行过程中不能死板地遵守初始值。比如一个任务本来预计消耗 50K Token，跑到一半发现外部数据比预期多，预计还需要 80K。如果死守 50K 就会强制终止，白白浪费前面已经消耗的部分。我在资源管控里增加了一个“动态再规划”机制：当剩余预算低于阈值的 20% 时，暂停执行，运行一个预算法则判断，是调高预算继续，还是进入降级模式。降级模式包括三种策略：摘要模式：把上下文历史压缩得更激进，减少每轮 context 的 token 消耗。探索限制：禁止调用高成本的探索性工具（比如网页搜索、长时间运算），只保留必需的数据查询和写操作。步骤精简：让模型用更少的中间步骤来完成任务，比如跳过不必要的验证环节，直接输出最终结果。动态调整的决策逻辑是“任务价值”优先。如果这是一个重要业务任务，预算翻倍也得上；如果只是数据补全的辅助任务，降级后跑完就算了，不值得追加资源。6.3 全局资源看板与熔断对运行中的 Agent，我还会在全局维护一个资源看板，内容包括：当前所有 Agent 实例的累计 Token 消耗、活跃并发数、平均响应延迟、最近 5 分钟 API 调用次数。看板数据通过一个独立线程定时采样，内存开销很小。熔断机制是资源管控的最后一道防线。当某个指标超过硬上限，比如单实例内存达到容器上限的 90%，或者 API 调用错误率超过 30%，触发熔断。熔断不是直接杀掉进程，而是“暂停-保存检查点-上报”。先把 Agent 冻在当前状态，保存检查点，然后向上层发出资源告警，等待人工决定是扩容后恢复还是中止任务。这种策略在实际运行里救了我很多次。有一次 Python 脚本不小心在循环里累积了巨大的中间对象，内存涨到快溢出，熔断机制把进程冻结并保存了检查点，重启后从检查点恢复，中间结果已经释放，顺利跑完。7. 一套完整流程串起来：23.3 版落地示例7.1 配置示例下面是一份典型的任务配置。我把这个文件命名为 task_config.yaml，放在任务目录下。它的作用不是让开发者手动配置所有细节，而是允许同一个运行器复用不同的任务模板。task: name: data_clean_and_cluster description: "读取 csv 文件，逐行清洗，最后做聚类分析" max_steps: 50 max_duration_sec: 1800 # 30 分钟硬限制 agent: model: "your-model-name" temperature: 0.2 system_prompt_file: "prompts/system.md" context: window_limit: 128000 # 模型窗口上限 reserved_for_output: 8000 # 输出预留 compress_threshold: 0.8 # 达到 80% 就压缩 max_recent_rounds: 8 # 保存最近 8 轮原始对话 summary_max_tokens: 4000 checkpoint: dir: "./checkpoints" format: "json" # json / msgpack keep_last: 5 save_on_stage: true save_before_side_effect: true resource: token_budget: 200000 api_call_budget: 100 max_concurrency: 2 memory_limit_mb: 2048 qps_limit: 5 auto_downgrade: true recovery: max_restarts: 3 verify_on_restart: true其中compress_threshold设成 0.8 表示当上下文估算 token 达到(window_limit - reserved_for_output) * 0.8时，启动压缩。max_recent_rounds控制保留最近 8 轮完整历史。资源预算里，auto_downgrade: true表示动态调整关闭时触发降级策略。7.2 核心代码骨架（Python）下面是一个精简但功能完整的运行器骨架。它把检查点加载、循环控制、资源监控、保存检查点这四件事串在一起。代码不能直接跑，但结构和关键逻辑可以照着改。import hashlib import json import os import time import uuid from dataclasses import dataclass, asdict @dataclass class CheckpointData: schema_version: int = 1 step_index: int = 0 agent_state: dict = None result_index: dict = None executed_actions: list = None context_summary: str = "" resource_meter: dict = None updated_at: float = 0.0 class AgentRuntime: def __init__(self, config, resource_monitor): self.config = config self.monitor = resource_monitor self.ckpt_path = config["checkpoint"]["dir"] os.makedirs(self.ckpt_path, exist_ok=True) def build_context(self, recent_messages, summary, new_step_info): ctx = [] if summary: ctx.append({"role": "system", "content": f"[历史摘要]\n{summary}"}) ctx.extend(recent_messages) ctx.append({"role": "user", "content": f"[当前步骤]\n{new_step_info}"}) return ctx def save_checkpoint(self, step_index, state, result_index, executed_actions, summary, meter, label="step"): data = CheckpointData( step_index=step_index, agent_state=state, result_index=result_index, executed_actions=executed_actions, context_summary=summary, resource_meter=meter, updated_at=time.time(), ) raw = json.dumps(asdict(data), ensure_ascii=False, indent=2).encode("utf-8") checksum = hashlib.sha256(raw).hexdigest() # 先写临时文件再 rename，避免写一半损坏 tmp_path = os.path.join(self.ckpt_path, f"ckpt_{label}.tmp") final_path = os.path.join(self.ckpt_path, f"ckpt_{step_index}_{label}.json") with open(tmp_path, "wb") as f: f.write(raw) with open(tmp_path, "a") as f: f.write(f"\n# checksum: {checksum}") os.rename(tmp_path, final_path) # 清理旧检查点，保留 keep_last 个 self._prune_old_checkpoints(final_path) return final_path def load_latest_checkpoint(self): files = sorted( [f for f in os.listdir(self.ckpt_path) if f.endswith(".json")], key=lambda f: os.path.getmtime(os.path.join(self.ckpt_path, f)), reverse=True, ) for fname in files: path = os.path.join(self.ckpt_path, fname) try: raw = open(path, "r", encoding="utf-8").read() body = raw.split("\n# checksum: ")[0] expect = raw.split("\n# checksum: ")[1].strip() actual = hashlib.sha256(body.encode("utf-8")).hexdigest() if actual != expect: print(f"[recovery] checksum mismatch: {fname}") continue ckpt = json.loads(body) print(f"[recovery] load success: {fname}") return ckpt except Exception as e: print(f"[recovery] load failed: {fname}: {e}") return None def run(self): ckpt = self.load_latest_checkpoint() if ckpt is not None: step_index = ckpt["step_index"] state = ckpt["agent_state"] result_index = ckpt["result_index"] or {} executed = ckpt["executed_actions"] or [] summary = ckpt["context_summary"] or "" meter = ckpt["resource_meter"] print(f"[run] resuming from step {step_index}") else: step_index = 0 state = {"initialized": True} result_index = {} executed = [] summary = "" meter = {"token_used": 0, "calls": 0} print("[run] fresh start") while step_index < self.config["task"]["max_steps"]: if self.monitor.is_budget_exhausted(): print("[run] resource budget exhausted, pausing...") self.save_checkpoint(step_index, state, result_index, executed, summary, meter, label="pause") break if self.monitor.overall_time_exceeded(): print("[run] time budget exceeded, pausing...") break # 组装上下文 recent_messages = self._load_recent_messages(state) step_info = f"第 {step_index} 步，当前任务目标..." ctx = self.build_context(recent_messages, summary, step_info) # 请模型决策（实际调用封装在 model_client 里） decision = self.model_client.decide(ctx) if decision.action == "tool_call": # 副作用工具调用前要检查幂等 action_id = uuid.uuid4().hex if decision.tool_name in result_index and \ result_index[decision.tool_name] == decision.tool_args: print(f"[run] duplicated call skipped: {decision.tool_name}") else: result = self.tool_executor.execute( decision.tool_name, decision.tool_args ) # 记录执行结果 result_index[f"result_{step_index}"] = { "tool": decision.tool_name, "args": decision.tool_args, "result_ref": self._store_result(result), } executed.append(action_id) # 更新上下文摘要 summary = self._update_summary(summary, decision, result) elif decision.action == "finish": print(f"[run] task finished at step {step_index}") self.save_checkpoint(step_index, state, result_index, executed, summary, meter, label="finish") return result_index step_index += 1 meter["calls"] += 1 meter["token_used"] += self.model_client.last_usage() # 循环边界保存检查点 if self.config["checkpoint"]["save_on_stage"]: self.save_checkpoint(step_index, state, result_index, executed, summary, meter, label="stage") print("[run] loop ended without finish")这段骨架我把执行过程简化了不少，但保留了对本文所有关键点的对应关系。你写自己的运行器时，重点也不在代码本身，而在里面那几个条件块：恢复前加载检查点、副作用调用前的幂等判断、预算耗尽时保存暂停检查点。这三条决定了你的 Agent 是不是真的“可恢复”。7.3 一次实测运行记录下面是一次用 23.3 跑真实任务的日志节选，我隐去了具体业务信息：[recovery] load success: ckpt_36_stage.json [run] resuming from step 36 [run] step 36: tool_call -> db_query, cost 1.2s, result 8 rows [run] step 37: tool_call -> external_classify, cost 3.0s, result ok [run] step 38: token_used=152400, remaining=47600, ratio=0.76 [context] compress triggered, summary token 3800 [run] step 39: tool_call -> dedup_merge, cost 0.8s, result 96 rows -> 82 rows [run] step 40: tool_call -> side_effect: send_notification, action_id pre-check skipped [run] step 41: finish, final result stored at result_index可以看到第 36 步是恢复点，说明之前中断过一次，恢复模块正常读档。第 38 步触发了上下文压缩，因为 token 占比到了 76%，压成摘要后继续跑。第 40 步调用了一个有副作用的通知接口，因为检查点里已经有该 action 的记录，所以直接跳过没有重复发送。这里我要说明，第 40 步的“跳过”说明副作用的幂等设计在工作，但也提醒我们：如果崩溃前通知已发出，但用户没收到（比如投递失败），那么跳过会导致漏发。所以对于关键副作用，最好在检查点里同时保存“响应状态”，只有状态为 success 时才跳过，状态为 unknown 的，应重新执行并带上幂等键。8. 常见问题与排查技巧实录我整理了几个高频问题，都是从实际运行里踩过的坑，格式做成速查表，方便你直接对照。现象可能原因排查思路解决与避坑恢复后 Agent 从头开始执行检查点没加载成功或恢复流程没提示模型看日志确认 load success；检查模型是否收到“从第 N 步继续”的指令在重建上下文中显式加入恢复指令，并把步骤游标写入系统提示恢复后重复发送邮件/写入数据副作用没有被幂等记录检查 executed_actions 是否完整；外部系统是否支持幂等键工具调用前先记录 action id；调用后把 state 写入检查点再真正执行上下文压缩后关键信息丢失摘要只记录了过程没记录结果检查摘要中是否包含“结果引用路径”摘要必须强制包含 result_index 中文件的存储路径循环卡在同一个工具调用模型没有从返回结果中获得有效信息对比最近 3 轮的工具参数是否一致开启发散检测，连续相同参数触发中断；把工具结果加工成模型更容易理解的结构检查点文件损坏直接覆盖写文件导致写了一半检查文件大小和校验和使用临时文件+rename；保留多个检查点运行到一半预算耗尽但任务还有价值预算初始值设定过死查看资源看板的剩余 token 和调用次数开启动态再规划，预算充足时扩容，不足时降级内存持续增长循环内中间数据没有释放观察进程 RSS 曲线；检查是否把大对象塞入上下文中间结果落盘，内存只保留引用；设置内存熔断恢复后上下文里全是旧历史，新信息加不进来摘要和原始上下文拼接顺序不对查看 build_context 拼接逻辑把摘要放在最前面，最近对话放在其后，当前步骤放在末尾除了表格里的问题，再分享两个排查习惯。第一个是给每个 Agent 任务打上 trace_id。检查点文件名、日志行、工具调用记录、资源看板全带同一个 trace_id。排查一次“队列里三个任务同时挂掉”的问题时，就是靠 trace_id 定位到某个共享连接池被耗尽导致的连锁失败。如果没有 trace_id，三个任务的日志混在一起，很难定位。第二个是恢复后做“检查点演练”。在绝不影响线上数据的前提下，故意把任务中断一次，然后立即恢复，检查恢复后状态是否正确。这个演练最好在开发环境跑通，否则生产环境第一次恢复时，你就只能看运气了。9. 个人经验与后续扩展我在实际应用这套机制时，最大的体会是把 Agent 当成一个有状态的服务来设计，而不是当成一个无状态的 API 调用。它和游戏存档很像：存档不需要记录你按下的每一次按键，只需要记录你当前在哪张地图、装备了什么、接了什么任务。检查点就是这套机制的“存档”，上下文就是你的“任务日志”，资源管控就是“货币和体力”。想清楚这个类比之后，整台运行机构的每个模块都变得很自然。另外我建议拿到这套代码后，先做一次“最小实现”：一个循环、一个检查点、一个恢复、一个资源计数器，跑一个只有 5 步的小任务，故意在第 3 步杀掉进程再恢复。把这套闭环跑通，再往里面加复杂的调度、多 Agent 协作、动态决策，都会顺很多。23.3 这套机制以后还可以扩展的方向，我目前在想的是把“检查点”本身变成一种可以搜索的状态库。也就是说，不只是最近的检查点，而是让某个 Agent 可以读取其他历史任务的检查点，复用它们的状态摘要和结果索引，减少从零开始规划的成本。另一个方向是针对多 Agent 协作场景做分布式事务，两个 Agent 共享一个任务状态时，检查点不能是单机文件，而要走分布式锁和提交日志。这个我还没完全跑通，等有结论了再来补充。总之，先把单 Agent 的这套机制吃透，比什么都重要。