复杂任务分解是现代 Agent 的必备能力

2026年6月，你在 Slack 里收到一条消息：“帮我整理 Q2 的销售数据，对比去年同期，生成一份简报，先发给团队 Review，通过后再同步到公司 Wiki。”这不再是一个简单的问答，而是一条包含数据提取、分析、格式化、审批流转、多渠道发布五个高阶目标的复杂指令。

如果你之前构建的 Skill 只具备记忆和单步工具调用能力，现在它会直接报错，或者更糟——假装成功、生成一份看似合理实则数据错误的报告。规划的缺失，是 Agent 从“玩具”到“工具”的第一道鸿沟。 这一章，我们要让 Skill 学会像项目经理一样思考：收到高阶目标后，自动拆解为可执行的步骤序列，执行中动态调整，并在失败时优雅回退。

你需要什么

最终成果

你将构建一个具备三层规划能力的 SalesReport Skill：

1. 层次化任务分解：输入“生成 Q2 销售简报”，自动输出一棵子任务树。
2. 动态重规划：当“提取数据”步骤失败时，不崩溃，而是尝试替代方案并重排剩余步骤。
3. 多 Agent 协同规划：将分析、审批、发布拆给三个虚拟子 Skill 并行执行。

做这个练习的原因：规划是 Agent 可靠性的基础。没有规划的 Agent 像没有 GPS 的司机——能开，但会迷路。

步骤说明

步骤 1：理解规划的本质——从目标到子目标的有向无环图

在动手写代码之前，先明确一个核心概念：任务分解的输出不是列表，而是有向无环图（DAG）。

为什么？因为真实任务的子任务之间往往存在“依赖”与“并行”两种关系：

生成 Q2 销售简报
├── 1. 提取 Q2 销售数据 (前置)
│   ├── 1a. 读取 CSV
│   └── 1b. 清洗空值
├── 2. 提取去年同期数据 (前置，与 1 并行)
├── 3. 计算同比变化 (依赖 1, 2)
├── 4. 生成图表 (依赖 3)
├── 5. 撰写文字简报 (依赖 3)
└── 6. 发送给团队审批 (依赖 4, 5)

列表无法表达“1 和 2 可以同时做，但 3 必须等它们都完成”。DAG 天然支持这种拓扑关系。

预期结果：你能画出一棵任务树，并用依赖关系判断哪些步骤可以并行。

在代码实现中，我们用一个 Plan 类来表示这种结构：

from dataclasses import dataclass, field
from typing import List, Dict, Optional

@dataclass
class SubTask:
    """子任务节点，支持依赖关系"""
    id: str                      # 唯一标识
    description: str             # 自然语言描述
    tool: Optional[str] = None   # 执行该子任务的工具名称
    depends_on: List[str] = field(default_factory=list)  # 依赖的子任务 ID
    status: str = "pending"      # pending | running | done | failed

@dataclass
class Plan:
    """任务分解计划，实质是一个 DAG"""
    goal: str
    subtasks: List[SubTask]

    def get_ready_tasks(self) -> List[SubTask]:
        """返回所有依赖已满足、可以执行的任务"""
        done = {t.id for t in self.subtasks if t.status == "done"}
        return [
            t for t in self.subtasks 
            if t.status == "pending" and set(t.depends_on).issubset(done)
        ]

步骤 2：让 LLM 生成结构化的分解计划

有了数据结构，接下来让 LLM 把自然语言目标转换成 Plan 实例。

设计规划提示词时有个关键原则：不要让 LLM 直接生成完整的动作序列，它只负责分解目标，每个子任务的具体执行由确定的工具完成。这与研究论文 TwoStep 的结论一致——LLM 擅长高层语义分解，但具体动作序列应由经典规划器或工具执行来保证正确性。

from deepagents import create_deep_agent  # 注意：不需要 langchain_ 前缀
from openai import OpenAI

client = OpenAI()  # 使用你的 API Key

# 规划系统提示——LLM 只做分解，不生成具体执行步骤
PLANNING_PROMPT = """你是一个任务规划专家。用户会给你一个高阶目标，你需要将其拆解为子任务。

要求：
1. 每个子任务只做一件事，可以映射到一个具体的工具调用。
2. 标注子任务之间的依赖关系（哪些必须先完成）。
3. 输出严格的 JSON 格式，不要给出具体的数据或执行结果——你只负责规划。

期待的 JSON 结构：
{
  "goal": "用户原始目标",
  "subtasks": [
    {
      "id": "1",
      "description": "从 sales 数据库提取 Q2 销售数据",
      "tool": "db_query",
      "depends_on": []
    },
    ...
  ]
}
"""

def generate_plan(goal: str) -> Plan:
    """调用 LLM 生成任务分解计划"""
    response = client.chat.completions.create(
        model="gpt-4o",  # 或 claude-3-5-sonnet 等
        messages=[
            {"role": "system", "content": PLANNING_PROMPT},
            {"role": "user", "content": goal}
        ],
        response_format={"type": "json_object"}
    )

    plan_dict = response.choices[0].message.content
    import json
    plan_data = json.loads(plan_dict)

    subtasks = [SubTask(**st) for st in plan_data["subtasks"]]
    return Plan(goal=plan_data["goal"], subtasks=subtasks)

# 测试
goal = "替我整理 Q2 销售数据，对比去年同期，生成图表简报并发送给 team@company.com"
plan = generate_plan(goal)

print(f"目标: {plan.goal}")
for t in plan.subtasks:
    dep_info = f"（依赖：{', '.join(t.depends_on)}）" if t.depends_on else ""
    print(f"  [{t.id}] {t.description} -> 工具: {t.tool} {dep_info}")

预期结果：LLM 返回一个结构化的 JSON，包含 5-7 个子任务，每个子任务都有明确的工具映射和依赖关系。关键观察是——LLM 没有生成具体数据，它只是“画了一张地图”。

踩坑提醒：如果你直接要求 LLM 输出包含具体执行步骤的完整计划，它会倾向“伪造”中间结果。例如在没有实际查询数据库之前就写出“Q2 收入增长了 12%”。永远把规划（What to do）与执行（How to do）解耦。

步骤 3：构建执行循环——按拓扑顺序逐个击破

有了 Plan，下一步是让 Agent 按依赖关系顺序执行。

import time

def execute_plan(plan: Plan, tool_registry: Dict[str, callable]) -> Plan:
    """
    按拓扑顺序执行所有子任务。tool_registry 将子任务的 tool 字段映射到实际函数。
    关键设计：get_ready_tasks() 从 DAG 中找出所有依赖已满足的任务——天然支持并行。
    """
    while True:
        ready_tasks = plan.get_ready_tasks()

        if not ready_tasks:
            # 检查是否全部完成
            all_done = all(t.status == "done" for t in plan.subtasks)
            any_failed = any(t.status == "failed" for t in plan.subtasks)
            if all_done:
                print("\n✅ 所有任务完成")
                break
            elif any_failed:
                print("\n⚠️  有任务失败，暂停执行")
                break
            else:
                # 理论上不会进入这里，除非有循环依赖
                print("\n⚠️  存在未解决的任务但无可用任务，检查循环依赖")
                break

        # 顺序执行就绪任务（可改为并行，但为演示清晰性用顺序）
        for task in ready_tasks:
            print(f"\n🔧 执行子任务 [{task.id}]: {task.description}")
            task.status = "running"

            tool_func = tool_registry.get(task.tool)
            if not tool_func:
                print(f"  ⚠️  未找到工具 '{task.tool}'，标记失败")
                task.status = "failed"
                continue

            try:
                result = tool_func()  # 实际场景会传入上下文
                task.status = "done"
                print(f"  ✅ 完成: {result[:60]}..." if len(str(result)) > 60 else f"  ✅ 完成: {result}")
            except Exception as e:
                task.status = "failed"
                print(f"  ❌ 失败: {e}")

        time.sleep(0.5)  # 模拟执行时间，实际中可移除

    return plan

# 模拟工具注册表
def mock_db_query():
    return "已提取 Q2 销售数据：总销售额 ¥2,340,000"

def mock_data_clean():
    return "数据清洗完成：移除 3 条异常记录"

def mock_calc_growth():
    return "同比增长 15.2%，环比增长 6.8%"

tool_registry = {
    "db_query": mock_db_query,
    "data_clean": mock_data_clean,
    "calc_growth": mock_calc_growth,
    # "send_email": lambda: 1/0  # 取消注释以测试失败回退
}

executed_plan = execute_plan(plan, tool_registry)

预期结果：你会看到子任务按依赖关系逐个执行，get_ready_tasks() 在每一轮只返回依赖已满足的任务。如果有两个任务都无前置依赖，它们会在同一轮被取出（目前顺序执行，但为并行留下接口）。

步骤 4：动态重规划——失败时不崩溃，而是寻找替代路径

上面的执行循环在遇到 "failed" 状态时会暂停。真实场景中，我们需要动态重规划：某一步失败后，基于当前状态重新生成剩余计划。

动态重规划的核心逻辑如下：

def replan_after_failure(plan: Plan, failed_task: SubTask) -> Plan:
    """
    当子任务失败时，尝试重新规划剩余未执行的任务。
    将已完成的任务写入上下文，告诉 LLM 当前状态，让它生成一个新计划。
    """
    # 找出所有已完成和待执行的任务
    done_tasks = [t for t in plan.subtasks if t.status == "done"]
    pending_tasks = [t for t in plan.subtasks if t.status == "pending"]

    # 构建当前状态描述
    done_desc = "\n".join([f"- [{t.id}] {t.description}: 已完成" for t in done_tasks])
    failed_desc = f"- [{failed_task.id}] {failed_task.description}: 执行失败"

    replan_prompt = f"""原计划执行过程中，以下步骤已完成或失败。请基于当前状态重新规划剩余步骤。

已完成步骤：
{done_desc}

失败步骤：
{failed_desc}

剩余待完成的步骤：
{pending_tasks}

原始目标：{plan.goal}

请生成一个新的计划（JSON 格式），只包含剩余需要执行的子任务。"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": PLANNING_PROMPT},
            {"role": "user", "content": replan_prompt}
        ],
        response_format={"type": "json_object"}
    )

    import json
    new_plan_data = json.loads(response.choices[0].message.content)
    new_subtasks = [SubTask(**st) for st in new_plan_data["subtasks"]]

    # 替换原计划中的失败任务及所有后续待执行任务
    # 注意：这里简化处理，实际需要处理 ID 冲突和已完成任务保留
    success_ids = {t.id for t in done_tasks}
    plan.subtasks = [t for t in plan.subtasks if t.id in success_ids] + new_subtasks
    plan.subtasks.sort(key=lambda t: (len(t.depends_on), t.id))  # 简单排序

    print("\n🔄 已动态重规划，新计划包含以下额外步骤：")
    for t in new_subtasks:
        print(f"  [{t.id}] {t.description}")

    return plan

注意：动态重规划的质量取决于 LLM 对失败原因的理解。如果“数据查询”失败可能是因为数据库不可用，LLM 可能会规划一个“从缓存读取”或“提示用户手动上传”的替代方案。你可以在 failed_task 中加入错误信息帮助 LLM 做出更好的判断。

预期结果：如果你取消注释 tool_registry 中的失败模拟，会看到执行暂停，动态重规划生成新计划，然后继续执行。

步骤 5：与多 Agent 系统的协同规划

当任务足够复杂时，单一 Agent 的单线程执行会变得缓慢。协同规划将总计划分配给多个专业化 Skills，实现协作执行。

在我们的示例中，三个虚拟子 Skill 各司其职：

利用 Deep Agents 内置的子代理生成（subagent‑spawning）机制，我们可以这样编排：

from deepagents import create_deep_agent

# 创建主协调 Agent
coordinator = create_deep_agent(
    model="gpt-4o",
    system_prompt="你是销售报告协调员，负责任务分解与子代理调度。",
    tools=[]  # 主协调员不直接执行，只做规划
)

# 定义三个专业化子 Agent（实际中它们会有各自的工具集）
data_fetcher = create_deep_agent(
    model="gpt-4o",
    system_prompt="你是数据提取专家，负责从各种数据源提取并清洗数据。",
    tools=[mock_db_query, mock_data_clean]
)

analytics = create_deep_agent(
    model="gpt-4o",
    system_prompt="你是数据分析师，负责计算指标并生成图表。",
    tools=[mock_calc_growth]
)

# 协同执行的关键：先由协调员分解，再将子任务分派给对应的子 Agent
def collaborative_execution(goal: str):
    """由协调 Agent 分解目标，并将子任务分派给专业化子 Agent"""

    # 第一步：协调员分解
    plan = generate_plan(goal)
    print(f"📋 协调员制定计划，共 {len(plan.subtasks)} 步\n")

    # 第二步：将子任务映射到子 Agent
    agent_map = {
        "db_query": data_fetcher,
        "data_clean": data_fetcher,
        "calc_growth": analytics,
        # "send_email": publisher  # 可继续扩展
    }

    # 第三步：按依赖关系执行（这里展示核心概念，省略完整实现）
    for task in plan.get_ready_tasks():
        assigned_agent = agent_map.get(task.tool)
        if assigned_agent:
            print(f"🤖 将 [{task.id}] 分配给 {assigned_agent.config.get('name', '子Agent')}")
            # 实际调用：assigned_agent.invoke(task.description)

    return plan

collaborative_execution(goal)

协同规划的关键原则：子 Agent 之间不应直接互相调用。所有调度逻辑由协调员集中管理，子 Agent 只负责执行分配到的原子任务。这避免了循环依赖和状态漂移。

预期结果：你会看到不同子任务被分配给不同的子 Agent。虽然这个示例是简化的，但结构可以扩展到真实场景，如 DataFetcher 连接数据库、AnalyticsEngine 调用绘图库、Publisher 操作邮件 API。

回顾

在 35-45 分钟内，你完成了以下事情：

1. 理解了任务分解的数据结构——用 DAG 而非列表表示计划，天然支持依赖与并行。
2. 实现了 LLM 驱动的计划生成——LLM 只做高层分解，不越界执行具体步骤。
3. 构建了拓扑顺序的执行循环——确保依赖关系严格遵守。
4. 加入了动态重规划能力——失败后基于当前状态重新生成剩余计划，而非直接崩溃。
5. 体验了多 Agent 协同规划——将总计划分派给专业化子 Skill。

行动清单

• [ ] 在你的项目中，找一个现有 Skill，为其添加 Plan 数据结构。
• [ ] 为关键流程编写规划提示词，确保 LLM 只分解目标不生成具体数据。
• [ ] 实现 get_ready_tasks() 和 replan_after_failure()，让执行循环具备容错性。
• [ ] 如果任务涉及多个独立步骤，考虑拆分为多个子 Agent 并行执行。

规划让 Agent 从“一次性回答”进化为“多步骤可靠执行”。但即便有了规划，执行过程中仍然会有各种意外：工具调用超时、API 返回格式错误、权限不足……真正的生产环境 Agent 必须有防御性编程和优雅降级的能力。

下一章《错误处理设计直接决定了 Skills 的生产可用性》将分析典型故障模式，教你如何设计重试、降级和告警机制，确保你的 Skill 不会静默失败——用户永远不该在三天后才发现机器人其实什么都没做。