别让Agent一口吃成胖子——从任务拆分到动态重规划的AI Agent工程化实战

xiaoxingkongyuxi

684人浏览 · 2026-09-05 07:00:00

xiaoxingkongyuxi · 2026-09-05 07:00:00 发布

AI Agent	任务拆分	大语言模型	LLM	Agentic Workflow
Plan-and-Execute	多智能体	工作流编排	可靠性工程	工程实践
摘要当 AI Agent 面对调研、分析、工具调用和报告生成等复合目标时，稳定性差距往往不在模型大小，而在任务能否被拆成边界清晰、依赖明确、可验证、可恢复的小闭环。本文讲解 Plan-and-Execute、DAG 并行、上下文隔离、动态重规划、幂等与回滚、人工审批、评估与可观测性，并给出 Planner-Executor-Evaluator 架构、Python 核心代码和端到端案例，回答拆到什么粒度、何时重试或重规划、哪些步骤可以并行，以及怎样把“能跑一次”的 Agent 升级为可控、可调试、可扩展的工程系统。

最危险的 Agent 指令，往往只有一句话

“帮我把数据库优化一下。”

这句话对人类工程师来说，只是一次需求澄清的开始；对一个拥有数据库、Shell、文件系统甚至发布权限的 Agent 来说，却可能被误解成一整套自作主张的行动：扫描表结构、修改索引、迁移数据、重启服务、清理历史表、写回配置。问题不在于模型“够不够聪明”，而在于系统把一个目标、几十个假设和多个高风险副作用一次性交给了同一个推理循环。

复杂 Agent 的事故链往往惊人地相似：目标太宽 → 模型自行补全隐含条件 → 上下文越滚越长 → 工具调用混在推理里 → 中间结果没有验收 → 某一步出错却无法局部恢复 → 最后只能整轮重跑，或者更糟，错误已经写进外部系统。

一句话结论 Agent 工程化的核心，不是把一个更长的 Prompt 塞给更强的模型，而是把复杂目标重构成一组“可执行、可验证、可恢复”的任务单元，并用明确的依赖和状态机把它们编排起来。

你会在这篇文章里解决什么

如何判断一个任务是否已经“小到可以稳定执行”，而不是把一句需求机械切成十句话。
如何把任务列表升级成 DAG：哪些步骤必须串行，哪些可以并行，哪里需要汇合与条件分支。
失败后到底应该 Retry、Replan、Rollback，还是立即进入 Human Gate。
如何让每个子任务只看到必要上下文，避免“全量历史”把模型注意力和 Token 预算一起拖垮。
如何给每一步定义验收条件、完成判定和观测指标，让“模型说完成了”不再等于“真的完成了”。
如何用一套最小 Planner–Executor–Evaluator 框架，把这些原则落成可扩展的工程代码。

01 为什么复杂任务一口吃下，稳定性会迅速恶化	09 生产架构：Planner–Executor–Evaluator 三层闭环
02 拆分不是切句子：定义“可执行任务单元”	10 验收与可观测：证明 Agent 真的完成了任务
03 从任务列表到任务图：依赖、并行、汇合与条件	11 端到端案例：竞争对手调研 Agent
04 四种编排模式：什么时候该用哪一种	12 可复用 Python 核心框架
05 拆到什么粒度：原子性不是越细越好	13 八个高频反模式：看起来聪明，实际上更脆
06 上下文工程：每一步只拿到“刚刚好”的信息	14 上线前检查清单
07 失败处理：Retry、Replan、Rollback 与 Human Gate	15 结语：把“智能”关进工程边界里
08 动态重规划：让计划在执行中保持有效

01 为什么复杂任务一口吃下，稳定性会迅速恶化

大模型会推理，不代表一个无限增长的推理循环天然适合复杂工程任务。

“一步完成”最诱人的地方，是代码看起来最少：一个系统提示词、一个工具列表、一个 while 循环，模型自己想、自己调工具、自己判断结束。但当任务同时包含检索、计算、事实核验、写操作、长文生成和多轮纠错时，这种结构会把原本可以隔离的问题耦合到同一条上下文里。

表象	真正的工程原因	直接后果	拆分后的修复方式
理解偏差	目标包含多个隐含子目标与前提	前几步走错，后面全部建立在错误假设上	先生成任务契约，再逐步执行并验收
上下文失焦	搜索原文、日志、代码、草稿混在同一窗口	关键信息被稀释，冲突指令增加	按步骤裁剪上下文，只传必要依赖产物
执行不可控	推理与高风险工具调用没有权限边界	误写、误删、越权操作	每步绑定工具策略、风险级别与审批门
失败难定位	没有清晰步骤与中间状态	只能看到最终失败，无法复现	每一步独立记录输入、输出、耗时、错误
成本膨胀	整轮失败只能全部重跑	Token、API 与时间重复消耗	局部 Retry / Replan，只恢复受影响分支
“完成”不可信	由同一个生成循环口头宣告完成	遗漏、幻觉或格式错误进入最终结果	外部验收器根据可观察结果判定

先把一个误区拆掉 任务拆分不是为了“让模型少想一点”，而是为了让每一次思考都发生在明确边界内：输入是什么、允许做什么、输出长什么样、什么叫完成、失败后怎么办。

复杂度真正从哪里来

复杂 Agent 的难度通常由四个维度共同决定：步骤数量、依赖密度、不确定性和副作用风险。步骤很多但完全确定的 ETL，适合静态工作流；步骤不多但需要开放式研究的任务，反而需要更强的规划与评估；一旦涉及外部写操作，风险会进一步放大。

不确定性	副作用风险	优先策略	典型任务
低	低	单 Agent 或静态顺序流	格式转换、字段提取、固定模板生成
高	低	规划 + 探索并行 + 汇总评估	研究、竞品分析、多来源事实核验
低	高	确定性工作流 + 权限门 + 回滚	数据库迁移、配置修改、批量写入
高	高	分阶段计划 + 最小权限 + 强验收 + 人工审批	发布、资金/权限变更、不可逆操作

02 拆分不是切句子：定义“可执行任务单元”

一个好子任务，应该像一个设计良好的函数：职责单一、签名明确、返回值可检查。

最常见的“假拆分”是把一句长需求改写成几个短句，却没有改变执行边界。例如“搜索资料 → 分析资料 → 写报告”看起来已经有三步，但“分析资料”仍然可能同时包含事实抽取、分类、计算、判断、引用选择和观点生成，失败后依旧无法知道问题在哪。

六个原子性测试

1. 单一职责：这一步能不能用一个动词准确描述？如果必须用“并且、同时、然后”才能说清，通常还太大。

2. 输入闭合：执行所需信息能否明确列出，而不是依赖“上文应该有”“模型自己知道”。

3. 输出契约：能否提前定义结构、字段、格式、数量、引用或状态变化。

4. 可验证：能否用程序规则、数据查询、测试、事实证据或独立评估器判断是否通过。

5. 可恢复：失败后能否单独重试、替换工具或重规划，而不必把所有上游步骤重做。

6. 副作用有界：写入、删除、发送、发布等动作能否被权限策略、幂等键、事务或人工审批约束。

把自然语言步骤升级成“任务契约”

真正可调度的步骤，不应只是一个字符串。最低限度需要包含目标、输入依赖、允许工具、输出模式、验收条件、失败策略和资源预算。下面这个 JSON 足够作为很多 Agent 编排器的起点。

一个可落地的任务契约示例

{
"id": "S4",
"goal": "将三路检索结果归一化为证据表",
"depends_on": ["S1", "S2", "S3"],
"inputs": ["official_docs", "pricing_pages", "user_feedback"],
"tool_policy": {
    "allow": ["text.parse", "deduplicate", "date.normalize"],
    "write_external": false
},
"output_schema": {
    "type": "evidence_list",
    "required": ["claim", "source", "published_at", "confidence"]
},
"acceptance": [
    "每条结论至少绑定一个可追溯来源",
    "重复证据已合并",
    "时间敏感数据包含日期"
],
"timeout_s": 90,
"max_retries": 2,
"on_failure": "replan_branch"
}

为什么结构化输出非常重要 只要中间产物仍是自由文本，下游步骤就会再次“理解一遍上游到底说了什么”。结构化产物把这种隐式推理变成显式接口，既降低歧义，也便于缓存、持久化和自动验收。

03 从任务列表到任务图：依赖、并行、汇合与条件

列表只告诉你先后顺序；任务图才能表达真实的工程关系。

图 1｜任务拆分后的核心不是“步骤清单”，而是带依赖、并行和汇合语义的任务图。

如果 S1、S2、S3 互不依赖，就没有理由让它们排队串行执行。真正应该串行的是依赖链；真正可以并行的是独立分支。对 I/O 型工具调用尤其如此：搜索、远程 API、文件读取往往等待时间远大于本地调度开销。

T_total ≈ T_plan + T_critical_path + T_merge + T_eval

这里的关键不是“把所有步骤都并发”，而是找到关键路径。任务图中的总耗时更接近最慢依赖链，而不是所有步骤耗时之和。并行越多，速率限制、共享状态竞争、结果合并和成本峰值也越明显，因此需要显式并发上限；工程上可以从 3～5 个并发任务起步，再根据 API 限流、CPU/内存、工具类型和失败率调优。

五种最常见的边

关系	含义	适用场景	工程要点
顺序	B 必须消费 A 的结果	先检索再分析、先编译再测试	上游失败时阻止下游进入 Ready
并行扇出	多个分支共享同一输入但互不依赖	多来源检索、多文件处理	限制并发；避免共享可变状态
汇合 Join	等待多个分支完成后再继续	证据归一化、结果聚合	定义“全到齐”还是“满足最小数量即可”
条件分支	根据中间结果走不同路径	是否需要补证、是否需要人工审批	条件必须可观测，不要只靠模型一句解释
循环	结果不合格时回到指定节点	修稿、测试修复、反思优化	必须设置最大轮次、预算或终止条件

并行的边界 只并行“读”和“计算”通常比较安全；对写数据库、发消息、改配置这类有副作用的动作，要谨慎并行，尤其不能让多个分支无协调地写同一资源。

04 四种编排模式：什么时候该用哪一种

不要为了“Agent 感”而动态规划。能确定的流程，固定下来反而更可靠。

模式	计划何时产生	优势	代价 / 风险	最适合
静态工作流	开发时固定	确定、便宜、容易测试	适应性弱	业务流程稳定、步骤可预测
ReAct	执行中逐步决定下一动作	灵活、实现简单	容易上下文膨胀，长任务全局性弱	短任务、工具选择少、探索范围有限
Plan-and-Execute	开始时先产出多步计划，再执行	全局结构清晰、便于依赖与并行	初始计划可能过时	中长任务、步骤较清楚但需要动态生成
Plan–Execute–Evaluate–Replan	计划与执行之间持续反馈	能局部修复、适合不确定环境	状态管理与评估成本更高	长任务、开放式研究、复杂工具链

一个稳健的选型原则是：先用最简单、最可控的结构满足任务；只有当任务确实需要开放式探索、不同专业角色或动态分支时，再增加规划器、多 Agent 或重规划环。复杂度本身不是能力，只有当它显著提高成功率、吞吐或可维护性时才值得引入。[1][2][6]

什么时候值得上多 Agent

需要同时探索多个彼此独立的信息空间，并行价值明确。
子任务需要明显不同的工具集、角色约束或专业知识，单一提示词会产生冲突。
上下文可以天然隔离，让子 Agent 只持有局部信息，并通过结构化产物交接。
存在清晰的协调器与终止条件，否则“多个 Agent 互相聊天”只会放大成本和不确定性。

反过来说 如果一个任务用单 Agent + 两三个工具就能稳定完成，先优化工具描述、输入契约和评估，而不是急着把它拆成“专家团队”。

05 拆到什么粒度：原子性不是越细越好

过粗会让失败不可控，过细会让调度和上下文交接吞掉收益。

“拆细一点”听上去总是安全，但过度拆分会带来新的问题：每一步都需要重新构造上下文、调用模型、记录状态和合并输出；大量微步骤还会制造不必要的依赖。理想粒度并不是最小步骤，而是“最小可独立验收单元”。

迹象	说明太粗	说明太细
职责	一个步骤同时做检索、分析、写作、发布	多个步骤只是同一个动作的机械切片
失败恢复	任何局部错误都要整步重跑	重试一个微步骤比直接重做更贵
输入输出	无法写清楚输入和输出格式	每一步产物都没有独立业务含义
验收	只能凭“整体看起来不错”判断	每一步都要做昂贵 LLM 评审
依赖	步骤内部隐藏了大量未建模前置条件	依赖图节点爆炸，调度开销占比过高

一个实用的粒度判断公式

拆分收益 ≈ 失败隔离收益 + 并行收益 + 验收收益 − 调度开销 − 上下文交接开销

这不是严格数学模型，但很适合作为评审时的思考框架。一个步骤如果既不能单独验证，也不能并行，也无法局部重试，拆出来往往只是“视觉上更整齐”；相反，一个步骤如果风险高、耗时长或不确定性大，即使只有一个动作，也值得单独成为节点。

06 上下文工程：每一步只拿到“刚刚好”的信息

上下文窗口是有限资源；高质量 Agent 不是“记住所有东西”，而是持续选择当前步骤真正需要的东西。

图 2｜把全量聊天历史替换为“任务上下文包”，让每一步只消费高信号信息。

长任务里最隐蔽的退化来自上下文污染：第一步的搜索结果、第二步的日志、第三步的草稿、第四步的错误堆栈不断累积，到后面模型虽然“看到了更多”，却更难判断什么仍然有效。工程上应把上下文当作可编排资源，而不是 append-only 日志。[3]

字段	包含什么	不要放什么
task_contract	当前步骤目标、输入、输出、验收条件	整份系统设计文档
dependency_artifacts	必要上游结构化产物或摘要	所有兄弟分支原文
tool_policy	允许工具、参数约束、读写权限	无关工具说明
working_state	当前事实、关键假设、短摘要	完整历史推理过程
budget	Token、超时、重试、并发限制	模糊的“尽量节省”
trace_refs	可追踪的来源 ID、文件 ID、记录 ID	不可定位的口头引用

把“记忆”分成三层，而不是一个无限增长的聊天记录

第一层是 Working Context，只服务当前步骤，允许频繁裁剪；第二层是 Artifact Store，保存证据表、代码补丁、查询结果、结构化摘要等可复用产物；第三层是 Durable Log，记录计划版本、状态迁移、工具调用与外部副作用。三层职责不同：前者追求高信号，后两者追求可恢复和可审计。

Working Context：短、临时、面向当前节点；节点结束即可重建。
Artifact Store：有稳定 ID 和 Schema，下游按引用取用，而不是复制整段原文。
Durable Log：只记录工程事实与状态变化，支持恢复、复盘和问题定位。

关键区别 “模型还能不能记住”不是可靠性指标；“进程重启后能不能从确定状态继续执行”才是。

07 失败处理：Retry、Replan、Rollback 与 Human Gate

失败处理策略如果只剩“再试一次”，复杂 Agent 很快就会进入重复犯错。

图 3｜同一个计划偶然失败用 Retry；计划本身失效用 Replan；高风险或责任边界不清时进入人工接管。

失败类型	典型信号	正确动作	不要做
瞬时工具故障	超时、429、网络抖动、临时 5xx	Retry：指数退避 + 抖动 + 上限	立刻改写整个计划
参数/格式错误	模式校验失败、缺字段、类型不符	局部修复参数或重新生成该步骤	把错误输出继续传下游
输入缺失	依赖产物为空、来源不可用	Replan：补充前置步骤或替换来源	无脑重复相同搜索
前提被推翻	新证据与计划假设冲突	Replan：更新任务图与受影响分支	假装旧计划仍然成立
外部写入失败	部分成功、重复写入风险	幂等检查 + 补偿 / 事务回滚	直接整轮重跑写操作
高风险不确定	删除、发布、权限、付款等	Human Gate：展示 diff、影响范围与证据	让模型自己判断“应该没事”

副作用操作必须多一层工程设计

幂等键：同一个任务重试不会产生第二次订单、第二条消息、第二次写入。
Dry-run / Preview：先生成变更计划、SQL diff、文件 diff 或待发送内容，再执行真实写操作。
最小权限：子任务只拿到它需要的工具和作用域，读工具与写工具分离。
补偿动作：无法使用事务时，为高价值写操作定义逆向恢复步骤。
人工审批：不可逆、跨权限边界、影响范围不确定时必须停下来确认。

可靠性底线 Agent 可以自主“推理”，但不能自主扩大“权限”。计划的动态性和权限的动态性是两回事；后者必须由策略层控制。

08 动态重规划：让计划在执行中保持有效

计划不是一次性文档，而是随新证据变化的运行时对象。

Plan-and-Execute 比一条无限 ReAct 链更有全局结构，但它仍然有一个现实问题：计划是在信息不完整时生成的。搜索可能找不到数据，API 可能返回新字段，测试可能揭示新的依赖，用户也可能在执行中补充约束。因此，生产系统需要把 Replan 当作显式能力，而不是靠模型“顺手改变主意”。

五类重规划触发器

1. 依赖失败：上游产物为空、不可访问或不满足质量阈值。

2. 新信息推翻假设：关键事实变化，导致后续步骤不再成立。

3. 验收失败：步骤已执行，但结果没有通过定义好的 acceptance criteria。

4. 预算失衡：某一分支持续消耗 Token / 时间，却没有获得足够信息增益。

5. 风险升级：从只读分析演变成写操作、发布或跨权限边界。

递归拆分要有停止条件

动态拆分最危险的不是拆不够，而是“越拆越多”。一个任务可以继续分裂成子任务、子子任务，如果没有终止规则，很容易形成规划递归。至少要同时设置深度、预算和最小任务价值三个限制。

停止条件	作用	建议实现
最大深度	阻止无限递归	max_depth / level
最小粒度	防止把一个简单动作继续切碎	满足原子性测试后禁止继续 split
预算上限	限制 Token、时间、工具调用和费用	budget_remaining <= threshold 时降级
信息增益阈值	阻止重复探索同一问题	连续 N 次没有新增有效证据则停止
终止条件	避免多 Agent 对话无休止继续	最大消息数、超时、显式完成信号

09 生产架构：Planner–Executor–Evaluator 三层闭环

把“脑、手、状态”解耦，才有机会让复杂 Agent 变得可维护。

图 4｜Planner 负责计划，Executor 负责受控执行，Evaluator 负责验收；状态仓库和工具层独立存在。

一个容易演进的 Agent 运行时，可以把职责拆成三层：Planner 只负责生成/更新任务图；Executor 只从 Ready 队列取可执行任务并调用工具；Evaluator 不参与“怎么做”，只根据任务契约检查结果是否通过。状态仓库贯穿三层，保存任务、产物、轨迹和副作用记录。长任务或多会话执行时，这种结构化交接尤其重要。[4][5]

Planner：不要输出漂亮计划，要输出可调度计划

步骤必须有稳定 ID，后续重规划才能精确替换节点。
depends_on 必须显式，不能把依赖藏在自然语言描述里。
每一步需要 acceptance 与 failure_policy，否则执行器不知道何时结束、失败后去哪。
计划要带资源预算：并发、超时、最大重试、最大深度。

Executor：像任务调度器，不像“自由发挥的聊天机器人”

只执行依赖已满足的 Ready 节点。
根据步骤风险绑定工具白名单和参数约束。
区分纯函数型步骤与有副作用步骤，后者必须支持幂等 / 补偿 / 审批。
输出必须写回结构化 artifact store，而不是只追加到对话。

Evaluator：验收结果，而不是听 Agent 自我汇报

优先使用确定性检查：JSON Schema、SQL 查询、文件存在性、单元测试、静态分析、数值阈值。
只有语义质量难以程序化时，才使用模型评审；评审标准必须具体，并保留被评内容和判定理由。
对事实性内容检查“证据是否支持结论”，不要只检查“语言是否流畅”。
评估器返回通过 / 不通过 / 需要补证，并指明受影响节点，给 Replan 提供结构化反馈。

10 验收与可观测：证明 Agent 真的完成了任务

“模型说完成了”只是一个文本事件；工程上需要可观察的结果状态。

Agent 的评估比普通单轮模型更难，因为一个任务可能跨多轮、调多个工具、修改外部状态，而且同一任务多次运行路径并不完全相同。评估的基本单位应该是“任务 + 运行轨迹 + 最终可观察结果”，而不是只看最后一段回答。[4]

每一步至少要有一个可机器检查的验收点

步骤类型	优先验收方式	示例
数据提取	Schema + 字段完整率 + 去重规则	required 字段 100% 存在；主键无重复
检索	来源数量 + 可访问性 + 时间范围	至少 3 个独立来源；关键 URL 可回溯
代码	测试 + 静态检查 + 运行结果	单元测试通过；退出码 0；无新增高危告警
数据库写入	事务状态 + 行数 + 业务约束	更新 37 行且无越界记录；可回滚点已建立
报告生成	结构规则 + 事实引用 + 语义评审	关键结论都有证据；章节齐全；无无来源数字

验收器也要分层：确定性检查优先，语义判断兜底

一个成熟的验收链通常不是“再调用一次 LLM 打分”，而是按成本和确定性分层。第一层使用代码型检查直接判定格式、状态、数量、测试结果和业务约束；第二层才使用模型评审处理事实支持度、覆盖度、表达一致性等语义问题；第三层保留给高风险或责任归属必须由人确认的场景。

代码验收：最快、可复现，适合 Schema、测试、数据库状态、文件和数值阈值。
模型验收：适合语义完整性、证据—结论一致性、复杂文本质量，但必须有明确 rubric。
人工验收：只放在不可逆副作用、高价值决策和自动规则无法可靠覆盖的边界上。

评估还要基于真实失败样本做回归：把曾经出现过的“来源缺失、工具超时、字段变化、错误写入、循环重试”等案例沉淀为固定任务集，持续观察升级模型、工具或提示词后是否引入新的退化。

生产环境值得长期看的一组指标

指标	你在观察什么	异常时通常说明什么
Task Success Rate	端到端任务真正通过验收的比例	整体能力或任务设计不够稳定
First-Pass Rate	无需 Retry / Replan 即通过的比例	计划质量、工具可靠性、提示词质量
Replan Rate	任务图被改写的频率	环境不确定性高，或 Planner 初始计划过于乐观
Retry Rate	同一节点重复执行的频率	工具抖动、参数生成不稳、超时设置不合理
p50 / p95 Latency	典型与尾部耗时	慢工具、长关键路径、并发受限
Token / Task	单任务模型上下文和生成成本	上下文污染、过度拆分、重复规划
Tool Error Rate	工具调用失败率	接口不稳、描述不清、权限/参数问题
Human Gate Rate	需要人工审批/接管的比例	高风险任务占比或自动验收能力不足

评估的目标 不是追求“每一步都让另一个模型打分”，而是尽可能把完成条件变成可执行断言。能用代码验证的，就不要交给主观判断。

11 端到端案例：竞争对手调研 Agent

用一个有检索、有并行、有失败分支、有语义评估的任务，把整套方法串起来。

图 5｜同一目标下允许局部失败和局部修复；重规划只改写受影响分支，而不是推倒重来。

假设用户提出：“分析 A、B、C 三个产品过去 12 个月的核心功能、价格变化、目标用户和公开口碑，给出可追溯证据，并生成一份 1500～2000 字的决策报告。”如果直接交给一个 Agent 一口气完成，它很容易出现来源混淆、时间错位、遗漏某家产品、数字无引用或在写作阶段重新编造事实。

第一步：先写任务契约，不急着搜索

字段	定义
目标	对 A / B / C 做同口径比较，结论服务于产品决策
时间范围	过去 12 个月，价格与版本变化必须带日期
来源优先级	官方文档 / 定价页 > 权威媒体 > 可验证用户反馈
输出	证据表 + 对比矩阵 + 关键洞察 + 风险 / 不确定项 + 最终报告
验收	三家覆盖完整；关键结论有来源；数字有日期；不确定信息明确标注
风险	全部为只读检索与本地分析，无外部写操作

第二步：Planner 生成任务图

ID	任务	依赖	可并行	验收条件	失败策略
S1	检索 A 的官方功能、价格与版本信息	S0	是	至少 2 类官方来源	换查询 / 替换来源
S2	检索 B 的官方功能、价格与版本信息	S0	是	至少 2 类官方来源	换查询 / 替换来源
S3	检索 C 的官方功能、价格与版本信息	S0	是	至少 2 类官方来源	换查询 / 替换来源
S4	补充公开口碑与第三方事实	S0	是	每家至少 2 条可追溯证据	缺哪家只补哪家
S5	归一化证据并去重	S1,S2,S3,S4	否	字段齐全、时间口径统一	修复格式 / 局部补证
S6	生成同口径对比矩阵	S5	否	三家都有值或明确 N/A	退回 S5
S7	生成洞察与风险	S6	否	每个结论映射到证据 ID	退回 S6 / 补证
S8	生成报告并验收	S7	否	结构、字数、引用、覆盖度全部通过	局部改写 / 回退对应节点

第三步：一个局部失败如何处理

S2 搜索 B 的历史定价时，目标页面返回 403。调度器不应该让整个任务失败，更不应该把 S1、S3 重新执行。首先判断这是“来源不可用”而不是“网络瞬时抖动”：如果重复请求仍失败，Replan 只替换 S2 的检索策略，例如改用官方帮助中心、版本公告、缓存页面或可信二级来源，并要求最终证据保留来源等级。新节点写回同一个 artifact slot，S5 无需知道上游具体换过几次策略。

第四步：Evaluator 不是只看文风

报告生成后，评估器发现一条结论“B 的价格在第三季度上涨”没有绑定任何证据 ID。此时最差的做法是让模型“润色一下整篇文章”；正确做法是返回结构化错误：missing_evidence: [claim_17]。Planner 根据 claim_17 反查来源节点，只补该证据或删除无法证明的结论，然后重新运行受影响的分析与报告节点。

这就是任务图的价值 同一个用户目标，在执行期间可以换搜索策略、替换失败节点、补充证据和回退局部结果；只要任务契约和产物接口稳定，系统不需要因为一个 403 或一条缺失引用从头重来。

12 可复用 Python 核心框架

下面是一套尽量小的骨架：任务图、Ready 调度、并发限制、Retry、验收和 Replan 钩子都在。

示例刻意不绑定任何模型厂商或 Agent 框架。你可以把 planner、execute_step、evaluate_step 换成自己的 LLM SDK、工作流引擎或内部工具层。重点不是代码长度，而是状态与职责边界。

核心调度器（1/3）：任务模型、状态与 Ready 判定

from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Awaitable, Callable
import asyncio, time

class Status(str, Enum):
  PENDING = "pending"
    RUNNING = "running"
    DONE = "done"
    FAILED = "failed"
    BLOCKED = "blocked"

@dataclass
class Step:
    id: str
    goal: str
    depends_on: list[str] = field(default_factory=list)
    acceptance: list[str] = field(default_factory=list)
    timeout_s: int = 60
    max_retries: int = 2
    risk: str = "read_only"     # read_only / reversible_write / irreversible
    status: Status = Status.PENDING
    output: Any = None
    error: str | None = None

@dataclass
class Plan:
    steps: dict[str, Step]
    version: int = 1

@dataclass
class EvalResult:
    passed: bool
    reason: str = ""
    replan: bool = False

ExecuteFn = Callable[[Step, dict[str, Any]], Awaitable[Any]]
EvalFn = Callable[[Step, Any], Awaitable[EvalResult]]
ReplanFn = Callable[[Plan, Step, EvalResult], Awaitable[Plan]]

class Orchestrator:
    def __init__(self, execute: ExecuteFn, evaluate: EvalFn,
                 replan: ReplanFn, concurrency: int = 4):
        self.execute = execute
        self.evaluate = evaluate
        self.replan = replan
        self.sem = asyncio.Semaphore(concurrency)
        self.artifacts: dict[str, Any] = {}

    def ready(self, plan: Plan) -> list[Step]:
        done = {sid for sid, s in plan.steps.items() if s.status == Status.DONE}
        return [
            s for s in plan.steps.values()
            if s.status == Status.PENDING
            and all(dep in done for dep in s.depends_on)
        ]

核心调度器（2/3）：单节点执行、超时、Retry 与验收

    async def run_step(self, step: Step) -> EvalResult:
        context = {dep: self.artifacts[dep] for dep in step.depends_on}
        last_error = None
        for attempt in range(step.max_retries + 1):
            try:
                step.status = Status.RUNNING
                async with self.sem:
                    result = await asyncio.wait_for(
                        self.execute(step, context), timeout=step.timeout_s
                    )
                verdict = await self.evaluate(step, result)
                if verdict.passed:
                    step.output = result
                    step.status = Status.DONE
                    self.artifacts[step.id] = result
                    return verdict
                if verdict.replan:
                    step.status = Status.FAILED
                    return verdict
                last_error = verdict.reason
            except Exception as exc:
                last_error = repr(exc)
                if attempt < step.max_retries:
                    await asyncio.sleep(min(2 ** attempt, 8))

        step.status = Status.FAILED
        step.error = last_error
        return EvalResult(False, last_error or "unknown error", replan=True)

核心调度器（3/3）：批量调度、Replan 与最终完成判定

    async def run(self, plan: Plan) -> dict[str, Any]:
        while True:
            pending = [s for s in plan.steps.values()
                       if s.status in {Status.PENDING, Status.RUNNING}]
            if not pending:
                break

            batch = self.ready(plan)
            if not batch:
                raise RuntimeError("没有 Ready 节点：可能存在失败依赖或循环依赖")

            verdicts = await asyncio.gather(
                *(self.run_step(step) for step in batch)
            )

            for step, verdict in zip(batch, verdicts):
                if not verdict.passed and verdict.replan:
                    plan = await self.replan(plan, step, verdict)
                    plan.version += 1
                    break

        failed = [s.id for s in plan.steps.values() if s.status == Status.FAILED]
        if failed:
            raise RuntimeError(f"任务未完成，失败节点: {failed}")
        return self.artifacts

执行层还要补上的生产能力

能力	为什么需要	最小实现
状态持久化	进程退出后继续任务	把 Plan、Step、Artifact 写入数据库或事件日志
幂等执行	重试不能重复副作用	task_id + step_id + operation 生成 idempotency key
超时与取消	慢工具不能拖死关键路径	asyncio timeout / cancel + 工具端 deadline
死循环检测	错误依赖或循环图会卡住	无 Ready 节点时 fail-fast；建图时做 DAG 校验
风险策略	不同步骤权限不同	risk → tool whitelist / approval policy
追踪	失败必须可复现	记录输入摘要、工具调用、产物 ID、耗时、错误、计划版本

不要让 Replan 悄悄修改已完成事实

重规划函数最容易踩的坑，是直接让模型返回一份“全新的计划”，把已经完成的节点、外部写入和审计记录全部覆盖。更安全的做法是让 Replan 输出 patch：新增哪些节点、废弃哪些未执行节点、更新哪些依赖、哪些已完成产物仍然有效。所有修改都带 plan_version，这样才能重放和审计。

推荐：让重规划输出增量 patch，而不是覆盖整份计划

{
"base_plan_version": 3,
"reason": "S2 官方历史价格页不可访问",
"add_steps": ["S2a_search_help_center", "S2b_search_release_notes"],
"disable_steps": ["S2"],
"rewire": {"S5.depends_on": ["S1", "S2a", "S2b", "S3", "S4"]},
"preserve_artifacts": ["S1", "S3", "S4"]
}

13 八个高频反模式：看起来聪明，实际上更脆

很多 Agent 失败并不是模型能力不够，而是架构把不确定性放大了。

反模式	为什么危险	替代方案
一切都交给一个超级 Prompt	目标、工具、限制和验收混在一起，失败不可定位	任务契约 + 分步执行 + 独立验收
拆得越细越专业	上下文交接和模型调用开销爆炸	以“最小可独立验收单元”为粒度
每个问题都上多 Agent	协调、终止、合并成本可能高于收益	单 Agent 稳定后再按专业/并行价值拆角色
把完整历史传给所有步骤	噪声与冲突持续累积	上下文包 + 结构化 artifact
失败统一重试三次	计划错误会被重复执行三遍	先分类：瞬时故障 Retry，计划失效 Replan
Evaluator 只问“是否完成”	模型容易被流畅答案说服	将验收拆成可执行断言 + 证据检查
写操作和读操作同权限	一次错误推理可能放大成外部事故	读写分离、最小权限、预览、人工门
没有终止与预算	循环、递归拆分、Agent 对话可能无限增长	最大轮次、Token、时间、深度、工具调用预算

14 上线前检查清单

如果下面大部分问题不能明确回答，先别急着给 Agent 更多工具权限。

☑ 每个步骤是否只有一个清晰职责？	☑ 输入依赖是否显式列出？
☑ 输出是否有结构化契约？	☑ 是否定义可验证的完成条件？
☑ 任务图是否经过循环依赖检查？	☑ 可并行节点是否限制并发？
☑ 是否区分只读与有副作用工具？	☑ 写操作是否具备幂等或补偿机制？
☑ 不可逆操作是否有人工审批？	☑ 是否设置超时、重试和退避？
☑ Retry 与 Replan 是否有不同触发条件？	☑ Replan 是否只修改受影响分支？
☑ 是否保留计划版本与审计轨迹？	☑ 上下文是否按步骤裁剪？
☑ 关键中间产物是否可持久化？	☑ 是否能从中断状态继续执行？
☑ 是否有端到端成功率指标？	☑ 是否统计 Retry / Replan / Tool Error？
☑ 是否观察 p95 延迟与 Token/Task？	☑ 是否准备真实失败样本做回归评估？

最后一道设计审查 不要问“这个 Agent 看起来聪不聪明”，要问：“如果第 7 步失败，我能不能只重做第 7 步？如果它要写外部系统，我能不能准确知道会改什么？如果它说完成，我能不能独立证明它真的完成？”

15 结语：把“智能”关进工程边界里

真正成熟的 Agent，不是一次做得更多，而是在复杂任务里始终知道自己正在做哪一步。

任务拆分的价值，不是把一个大问题变成一堆小 Prompt，而是建立一套可执行的“任务协议”：明确目标、显式依赖、结构化产物、最小上下文、受控工具、独立验收和失败恢复。做到这一步，Agent 才从一个长对话，变成了可以调度、测试、观测和演进的软件系统。

对确定性强的部分，用工作流锁住；对开放探索的部分，用 Planner 提供结构；对可并行的部分，用任务图缩短关键路径；对不确定结果，用 Evaluator 给出证据化验收；对变化中的环境，用 Replan 只修正受影响分支；对高风险副作用，用权限与人工门把模型能力限制在安全边界内。

记住这条主线 先拆边界，再建依赖；先定义验收，再允许执行；先保证局部可恢复，再追求全局自治。一个真正可靠的 Agent，不需要“一口吃成胖子”，它只需要把每一口都吃明白。

参考资料延伸阅读

以下资料用于进一步理解 Agent 工作流、上下文工程、评估与多智能体编排。

1. Anthropic — Building Effective AI Agents

2. Anthropic — How we built our multi-agent research system

3. Anthropic — Effective context engineering for AI agents

4. Anthropic — Demystifying evals for AI agents

5. Anthropic — Harness design for long-running application development

6. Microsoft AutoGen — AgentChat Teams / GraphFlow / Termination

7. OpenAI Agents SDK — Handoffs

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

程序员学 AI（八）：AI Agent 到底是什么？从回答问题到完成任务

AI Agent技术社区

22道AI Agent 工程师必会的面试题，别啥都不会就出去找工作

AI Agent技术社区

写给 TypeScript 工程师的 OpenClaw.NET 上手指南：用你熟悉的 TS 思维，跑起一个生产级 AI Agent

读代码层：TS 和 C# 本来就是「远房亲戚」——asyncawait??、接口、泛型、严格空检查，你的心智模型几乎无缝平移。真正要适应的只有两件事：类型是具名而非结构化的（多一句显式声明），以及 NativeAOT 下的「禁反射」约束。写扩展层：你甚至可以不怎么碰 C#——Node.js 插件桥意味着你的 TypeScript 插件就是一等公民，而且能在生产 aot 车道跑。内核给你 .NET

AI Agent技术社区

所有评论(0)

查看更多评论

xiaoxingkongyuxi

@xiaoxingkongyuxi

已为社区贡献118条内容

别让Agent一口吃成胖子——从任务拆分到动态重规划的AI Agent工程化实战

xiaoxingkongyuxi

你会在这篇文章里解决什么

目录

01 为什么复杂任务一口吃下，稳定性会迅速恶化

复杂度真正从哪里来

02 拆分不是切句子：定义“可执行任务单元”

六个原子性测试

把自然语言步骤升级成“任务契约”

03 从任务列表到任务图：依赖、并行、汇合与条件

五种最常见的边

04 四种编排模式：什么时候该用哪一种

什么时候值得上多 Agent

05 拆到什么粒度：原子性不是越细越好

一个实用的粒度判断公式

06 上下文工程：每一步只拿到“刚刚好”的信息

推荐的上下文包结构

把“记忆”分成三层，而不是一个无限增长的聊天记录

07 失败处理：Retry、Replan、Rollback 与 Human Gate

副作用操作必须多一层工程设计

08 动态重规划：让计划在执行中保持有效

五类重规划触发器

递归拆分要有停止条件

09 生产架构：Planner–Executor–Evaluator 三层闭环

Planner：不要输出漂亮计划，要输出可调度计划

Executor：像任务调度器，不像“自由发挥的聊天机器人”

Evaluator：验收结果，而不是听 Agent 自我汇报

10 验收与可观测：证明 Agent 真的完成了任务

每一步至少要有一个可机器检查的验收点

验收器也要分层：确定性检查优先，语义判断兜底

生产环境值得长期看的一组指标

11 端到端案例：竞争对手调研 Agent

第一步：先写任务契约，不急着搜索

第二步：Planner 生成任务图

第三步：一个局部失败如何处理

第四步：Evaluator 不是只看文风

12 可复用 Python 核心框架

执行层还要补上的生产能力

不要让 Replan 悄悄修改已完成事实

13 八个高频反模式：看起来聪明，实际上更脆

14 上线前检查清单

15 结语：把“智能”关进工程边界里

参考资料 延伸阅读

所有评论(0)

温馨提示：您尚未绑定手机号

xiaoxingkongyuxi

参考资料延伸阅读