多Agent任务分解框架：从模糊目标到可执行子任务清单

把一个大目标直接丢给单个 Agent，看起来省事，真跑起来就会现原形：要么输出一份四平八稳、但哪一步都落不了地的方案，要么写到一半开始重复套话，把开头提过的约束忘得干干净净。后来我把 Multi-Agent 任务分解框架放到中间，先让系统把目标拆成子任务，生成一张真正可执行的清单，再让不同角色照着清单干活，整个流程才稳定下来。

这篇文章就按这条链路来讲：为什么需要任务分解、怎么把一个模糊目标拆成结构化的子任务清单、子任务怎么编排给多个 Agent 执行，以及落地时那些容易被忽略的关键细节。最后我会给一个最小可跑的管线设计参考，方便你直接照搬改造。适合正在搭复杂 AI 应用、或者想把多个大模型调用串成流水线的开发者，也适合想弄明白“多个 Agent 到底怎么协作”的产品和技术负责人。

1. 为什么 Multi-Agent 系统离不开任务分解层

1.1 单个 Agent 处理复杂目标的真实瓶颈

先聊一个很现实的体验。我最早做复杂任务时，习惯把整个目标直接写进 Prompt，让模型一口气输出完整结果。例如要一份“某行业 2024 年市场分析与进入策略”，模型确实会给出看起来很完整的报告，但细看全是问题：前面两个章节还算有信息量，到后面就开始用一些正确的废话填充，甚至会把摘要里已经明确排除的范围又写回来。

这其实是大模型的固有特性造成的，不是模型不聪明。上下文窗口虽然越来越大，但长输入、长输出下注意力会稀释，模型很难在一万 token 的内容里始终保持初始约束。其次，“一次生成”没有中间锚点，模型只能用一次脑补完成从目标到结论的全过程，中间没有检查点，没有分阶段的验证标准，出错后只能整段重来。

更麻烦的是不可验证。你拿到一份生成好的分析报告，很难判断“它漏没漏关键维度”，因为整个过程是一个黑盒。如果把任务拆成独立的小块，每一块都能单独被检查和重新生成，问题就变成了可以逐步定位和修复的流程问题。

1.2 Multi-Agent 不等于“开多个会话”

很多人对多 Agent 的第一反应是：开几个对话窗口，同时问同一个问题，然后投票。这样做的结果往往是多个版本互相矛盾，你也不知道该信谁，本质上只是把同一个任务的随机性放大了。

另一种常见的做法是让多个 Agent 在同一个长对话里“聊天式协作”，A 说一句，B 回一句。在演示里效果很好，但一旦任务稍微复杂，对话就会陷入重复讨论和相互否定，而且每轮都要重传上下文，token 消耗直线上升。最要命的是没有明确的任务边界，谁负责哪一块始终说不清。

我理解的 Multi-Agent 不是让一堆模型闲聊，而是让它们像流水线上的工位一样协作。每个工位只处理一道工序，工件通过传送带传递。任务分解框架就是这个传送带：把目标切成若干子任务，定义好先后顺序和数据流，再按清单派发给不同 Agent。没有这层框架，多 Agent 就只是多个单 Agent 的堆叠，协作效率甚至会低于单 Agent。

1.3 任务分解框架解决的三类问题

第一个问题是大目标超出单次推理的合理范围。拆解后，每个子任务都可以设计成“一次模型调用能完成”的规模，输出质量会明显提升。

第二个问题是上下文污染。多 Agent 协作时，如果每个 Agent 都拿到完整的历史对话，无关信息会成为干扰，甚至让模型把别处的内容误当成自己的输入。通过清单传递数据，每个 Agent 只看到自己需要的那部分，大幅减少这种噪声。

第三个问题是局部重试成为可能。单 Agent 一次生成失败，往往要整锅重来；任务分解后，哪个子任务验收不通过就重跑哪个，依赖它的下游可以等待或改道，流程韧性完全不一样。这也是“可执行清单”这三个字的分量所在：能单独验证、能局部修复，才算真正可执行。

1.4 从目标到可执行清单的完整链路

整个链路我通常拆成五步：目标输入、范围分析、子任务生成、依赖建模、编排执行。目标输入是用户那句原始描述，往往模糊且充满隐含约束；范围分析先把这些约束提炼出来，确定什么在范围内、什么不在；子任务生成基于范围拆出任务列表；依赖建模标记任务之间的先后与并行关系；最后编排执行，调度 Agent 按清单干活并汇总结果。

这五步可以看作一句话：先拆再排，排完再派。前两步是“理解目标”，中间两步是“设计清单”，最后一步是“调度执行”。很多人只盯着最后一步，结果前面没拆好，调度再漂亮也没有用。下面我把每一步的关键方法展开讲。

2. 从模糊目标到结构化子任务的拆解方法

2.1 先翻译目标，再开始拆解

拆解的第一步不是直接列任务，而是先把目标“翻译”成一份结构化的目标描述。我会让 Planner 按一个固定模板输出目标解析结果，Prompt 大致是这样：

你是一个目标分析器。请把用户目标解析为 JSON，不要急着拆任务。 { "goal": "用户原始目标", "intent": "剥离表象后的真实意图", "scope": "需要覆盖的范围，尽量明确列举", "constraints": ["时间约束", "成本约束", "质量约束", "其他明确限制"], "acceptance_criteria": ["可被客观验证的完成条件，至少三条"], "risks": ["执行过程中最可能翻车的地方"] }

为什么要多做这一步？因为用户的目标通常是模糊的，比如“帮我做个市场分析”，这里没说要分析哪些维度、覆盖多长周期、给谁看。如果直接拆任务，Planner 只能靠猜，拆出来的清单大概率跑偏。先做一次目标澄清，相当于给后面的所有子任务定了一个锚点。

如果目标本身缺信息，我一般让 Planner 输出“假设的范围说明”，而不是停下来反问用户。实际产品里没有那么多来回对话的机会，先把假设写清楚，后续按这个假设执行，最后在报告里标注假设，比反复追问更实用。

2.2 拆解粒度怎么控制

拆解粒度是任务分解中最难拿捏的部分，我见过两个极端。一个极端是任务过粗，比如“写一份市场分析报告”直接作为一个子任务，这等于把原来的大目标换个名字又丢回给 Agent；另一个极端是任务过碎，比如“查 2021 年的市场规模”“查 2022 年的市场规模”“查 2023 年的市场规模”各拆成一个任务，调度开销大，每个任务还要重复读背景信息，token 全浪费在上下文上。

我的判断标准是四条：子任务有明确的输入来源、有可验证的输出格式、执行者只需要完成一件事、能在一次或至多两次模型调用内产出。按这个标准，一个“汇总三年数据”的任务比“查 2021 年”“查 2022 年”“查 2023 年”三个任务更合适，因为它把重复背景读取压缩成了一次，输出也更集中。

2.3 依赖关系建模

任务拆好之后，下一步是标记它们之间的关系。最常见的是串行依赖：A 的产出是 B 的输入，比如先定主题，再设计议程。第二种是并行：两个任务互不依赖，比如“列物资清单”和“估算天气风险”可以同时跑。第三种是条件分支，需要根据某个任务的输出决定下一步走哪条路，比如“如果调研发现目标人群偏好线上，就生成线上方案，否则生成线下方案”。第四种是汇总，多个并行任务的结果合并成一个输入，给最终的组装任务使用。

实践中我用一个简单的字段来建模：depends_on。每个任务记录它依赖哪些任务 ID，这样整个任务集合就构成一个 DAG（有向无环图）。调度时从入度为 0 的任务开始执行，完成一个就释放一批下游任务，这本质上就是拓扑排序。

这里有一个值得强调的点：依赖关系别只看“数据上需要”，还要考虑“业务上必须”。比如预算估算依赖物资清单，这是数据依赖，跳过中间步骤一定会出错；但文案生成依赖预算，更多是业务上希望文案里能带上预算亮点，如果预算任务失败，文案可以先出草稿再补数字。梳理依赖时我习惯先标数据强依赖，再标业务弱依赖，弱依赖可以设置为“可降级”。

2.4 可执行清单条目怎么写

一份可执行清单里的每个条目，不是一句话描述就完了。我通常要求包含这些字段：

举一个例子，目标“针对某产品做一份市场调研”，拆出来的片段可能是：

这里的关键是每条任务的描述里要写清楚“输出长什么样”，最好直接给出 JSON 结构或 Markdown 模板。含糊的“写一份分析”等于让 Agent 自己决定输出格式，后续程序根本没法统一解析和验收。

3. 子任务编排与多 Agent 协作执行

3.1 角色设计决定输出边界

任务清单里的agent_role不是装饰。角色本身的系统提示词，在很大程度上决定了 Agent 会关注什么、不关注什么。例如调研 Agent 的系统提示词里应该写明“只负责收集和整理事实，不要给出结论性判断”；分析 Agent 则写明“必须在结论后附上支撑证据，证据引用自输入数据”。

角色指令写得好不好，直接决定下游任务拿到的数据质量。我见过很多项目里调研 Agent 顺手写了结论，分析 Agent 又重复做了一遍调研，两边口径不一致，最后组装的人根本不知道该信谁。所以角色边界要写“硬”，不要让 Agent 自由发挥超出职责。

3.2 上下文传递：一条任务只喂它需要的输入

多 Agent 协作时，上下文传递方式我踩过一个大坑：最初为了方便，我把所有历史输出全部塞给每个 Agent，结果一个 Agent 拿到了另一个任务的中间数据，误把自己的输出格式改成别人的格式，整个下游解析都崩了。

后来改成“最小上下文”原则：构建某个任务的 Prompt 时，只从上下文池里取input_refs引用的那些输出。其他历史一概不传。这样做有两个直接好处：一是省 token，任务越多省得越明显；二是减少干扰，Agent 不会被无关信息带偏。

小小的伪代码示意：

def build_task_prompt(task, context_pool): inputs = {ref: context_pool[ref] for ref in task.input_refs} system = f"你是{task.agent_role}，请严格按描述输出。" user = f"你的输入数据：\n{json.dumps(inputs, ensure_ascii=False)}\n任务要求：{task.description}" return system, user

这里的context_pool就像一个消息总线，每个任务完成后把输出放进去，下游按需取用。这样即使某个上游失败，我们也能明确知道哪些任务会拿不到数据，而不是静默出错。

3.3 并行执行与结果合并

任务之间没有依赖关系时，可以并行执行。大模型 API 调用大多是网络 IO 密集型，用线程池就够了，不需要上太复杂的方案。一个简化的调度循环可以是：

from concurrent.futures import ThreadPoolExecutor, as_completed def execute_tasks(tasks, context_pool, model_call, max_workers=4): done = set() while True: ready = [ t for t in tasks if t.status == "pending" and set(t.depends_on).issubset(done) ] if not ready: break with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = { executor.submit(run_one_task, t, context_pool, model_call): t for t in ready } for future in as_completed(futures): task = futures[future] if future.exception() is None: task.status = "done" context_pool[task.task_id] = task.output else: task.status = "failed" done.add(task.task_id)

并行执行后，结果合并往往比想象中麻烦。不同 Agent 输出的术语口径不一致是最常见的：一个写“市场规模”，另一个写“销售额”，组装任务需要额外做一次统一。我的做法是增加一个“组装 Agent”，专门负责把各模块输出按最终模板合并，并处理明显的口径差异，但原则是不改动原始数据，只在最终报告中做映射。

3.4 子任务失败时的应对策略

再完善的清单也会遇到子任务失败。我的处理策略分三层。第一层是验收不通过时重试，重试时把上一次输出的错误信息作为反馈，让模型知道哪里不合格，最多重试两到三次。第二层是连续失败则标记failed，依赖它的任务直接标记skipped，避免下游拿着残缺数据继续跑出一堆看似正常但实际不可信的成果。第三层是对失败任务做归因，是 Prompt 不够清楚，还是输入数据不足，还是模型能力确实达不到，然后把失败任务的内容整理进最终报告的风险说明里。

这里最忌讳的是“让下游硬跑”。下游拿到了缺失数据，模型往往会自己编造一部分来补全，最后整篇结果看起来完整，实际上有一半是幻觉。宁可让下游跳过并明确标注，也不能让错误被静默掩盖。

4. 定义“可执行清单”时最容易被忽略的细节

4.1 验收标准必须能被程序检查

很多人在写子任务验收标准时，会写“输出要合理”“结论要有依据”。这类描述人眼能判断，程序完全没办法自动判定，任务状态只能靠人工维护，分分钟变成瓶颈。我后来强制要求：每个子任务的验收规则必须是可以被代码检查的规则。

比较实用的校验类型包括：输出是合法 JSON、必须包含某些字段、字段值类型正确、文本不低于某个长度、包含或排除某些关键词、数值在一定区间等。一个简化的校验函数长这样：

def check_acceptance(output, rule): if rule["type"] == "json": data = json.loads(output) return all(key in data for key in rule["required_keys"]) if rule["type"] == "min_length": return len(output) >= rule["min"] if rule["type"] == "no_forbidden": return not any(word in output for word in rule["forbidden"]) return True

能程序校验的子任务，才能真正放进自动流水线里跑。否则所谓“可执行清单”只是给人看的，离自动化还差一层。

4.2 数据契约：先统一字段名，再谈执行

多个子任务之间通过上下文池传递数据，字段名的一致性就成了硬约束。上游任务输出total_revenue，下游任务描述里写“总收入”，模型可能不知道这两个是同一个东西，尤其是输出格式不固定时，下游很容易取错字段。

解决方法是让 Planner 在拆解阶段就为每个任务定义输出 schema，而不是只写一段文字说明。比如 T2 的输出 schema 明确要求：

{ "type": "object", "properties": { "key_findings": {"type": "array"}, "source_count": {"type": "number"} }, "required": ["key_findings", "source_count"] }

下游任务再用到这些数据时，Prompt 里直接写“取key_findings字段”，一致性就保证了。这个环节有点像一个团队先定了 API 接口再各自开发，前后联调时才不会互相等。

4.3 幂等与缓存：别让随机性干扰调试

大模型天然有随机性，同样一个子任务连续跑两次，结果可能不同。这在单次使用中无所谓，但一旦进入流水线调试阶段，随机性就是灾难：一个任务失败后重试成功，你会分不清是修复生效了还是运气好。

我的经验是：调试阶段把模型的采样参数调到接近确定性的档位，同时对每个任务的输入加缓存。具体做法是把 Prompt 内容哈希后作为缓存 key，命中缓存就直接返回旧结果。这样同一个任务反复调试不会反复烧钱，出了问题也能复现。线上正式运行时再决定要不要放开随机性，但不管怎样，每个任务的输入输出和耗时都要打日志，否则出问题根本无从下手。

4.4 成本与超时：并行不是免费的

多 Agent 并行执行看起来很高效，但还有一个更容易被忽视的问题：成本是叠加的。四个任务并行跑，token 消耗就是四个任务同时计费，而一个超时的调用还会占住线程池里的一个位置，拖慢整条流水线。我通常在配置里设置一批默认值：

“能并行”不代表“应该并行”。如果任务之间共享大量背景，串行执行可以让模型把上下文整理得更完整，并行反而让每个任务都要重复读一遍相同背景。成本敏感时，优先保关键路径上的并行，非关键任务串行也不迟。

5. 从零搭一个最小可用的任务分解管线设计参考

5.1 先定义核心数据结构

我从第一版代码走到现在，最深的体会是：任务分解框架的根基不是 Prompt，而是数据结构。数据结构把“目标、子任务、依赖、验收”这些概念固定下来，后面的 Planner、执行器都是围绕它转的。最小可用版可以这样定义：

from dataclasses import dataclass, field from typing import List, Dict, Any @dataclass class Task: task_id: str title: str description: str input_refs: List[str] # 引用其他任务ID或初始输入 agent_role: str # 谁来执行 acceptance_rule: Dict[str, Any] # 程序可校验的规则 depends_on: List[str] = field(default_factory=list) status: str = "pending" # pending/running/done/failed/skipped output: str = "" # 执行结果 @dataclass class GoalPlan: goal: str scope: str constraints: List[str] tasks: List[Task]

我把Task设计成有状态的对象，是为了让执行器能随时知道整个流水线卡在哪个节点。调试时打印一份任务状态表，哪里失败一目了然。

5.2 目标解析与任务生成模块

Planner 的核心逻辑不难：一次模型调用，输入用户目标，输出包含任务列表的 JSON。难的是输出格式一点都不能错。我的做法是在系统提示词里给一个非常具体的输出模板，并要求“只输出 JSON，不要输出任何解释文字”。

def parse_goal(user_goal: str, model_call) -> GoalPlan: prompt = BUILD_PLANNER_PROMPT + user_goal raw = model_call(prompt) try: data = json.loads(raw) except json.JSONDecodeError: # 修复轮：把错误信息回传，让模型重新输出 raw = model_call(prompt + "\n你刚才的输出不是合法JSON，请只输出合法JSON。") data = json.loads(raw) tasks = [Task(**t) for t in data["tasks"]] return GoalPlan( goal=data["goal"], scope=data["scope"], constraints=data.get("constraints", []), tasks=tasks, )

修复轮是个实用技巧。模型偶尔会把 JSON 包进 Markdown 代码块里，或者带一段多余解释，这一步能救回大量解析失败。

5.3 调度执行器的核心循环

执行逻辑在前面已经拆过一遍，这里把完整流程串起来：循环扫描所有任务，选择依赖已完成且自身 pending 的任务，并发执行，校验产出，按结果更新状态。执行完成后，done任务的输出存进 context_pool，failed任务留给下游 skipped，最后把整个计划的状态和输出返回。

def execute_plan(plan: GoalPlan, model_call, max_retries=2, max_workers=4): pool = {"goal": plan.goal} with ThreadPoolExecutor(max_workers=max_workers) as executor: while True: ready = [ t for t in plan.tasks if t.status == "pending" and set(t.depends_on).issubset(pool) ] if not ready: break futures = { executor.submit(run_task_with_retry, t, pool, model_call, max_retries): t for t in ready } for future in as_completed(futures): task = futures[future] pool[task.task_id] = task.output return plan

这里run_task_with_retry内部就是前面提到的“构建 Prompt → 调用模型 → 校验验收规则 → 失败重试”逻辑。执行器不关心每个子任务具体写什么，只负责把数据喂进去、把结果存回来，职责边界非常干净。

5.4 一个完整例子：活动策划任务的拆解与执行

用“策划一场面向开发者群体的线下技术沙龙，包含主题、议程、物资和预算”这个目标来演示，Planner 拆出来的清单可能是：

实际执行时，T1 先跑；T2 等 T1 完成；T3 和 T4 都依赖 T2，可以在 T2 后并行跑；T5 最后汇总。整个流程符合“串行 + 并行 + 汇总”的组合，也不需要写复杂的调度器，按照depends_on扫描 ready 任务就能跑通。

5.5 什么场景适合这套框架，什么场景别硬上

最后说点经验之谈。任务分解框架不是银弹，我用下来的判断标准是：任务足够复杂、结果需要多角色协作、子任务之间可以定义清晰的输入输出边界。比如研究型报告生成、活动策划、产品方案设计、代码模块拆解，这类场景非常适合。

反过来，简单问答、闲聊、单一文档生成，硬套框架反而更贵更慢。拆解本身要消耗一次甚至几次模型调用，调度和验收也要代码成本，如果任务一次就能完成，直接调用单个 Agent 才是合理选择。我的体会是，这套框架真正的价值在于让复杂任务的执行过程变得可追踪、可重试、可优化。与其把多个 Agent 当聊天伙伴，不如把它们当成流水线上的工序，清单就是流水线上的工单。