AI Agent Task Decomposition 与 Planning 系统深度实践：从分层规划到自适应执行的全栈指南 🎯🧠

发布日期：2026-06-28

概述

在 2026 年的 AI Agent 工程实践中，任务分解与规划能力是区分简单聊天机器人与真正智能助手的关键分水岭。当 Agent 面对"帮我分析这份财报并生成投资建议报告"这样的复合指令时，能否自主将目标拆解为可执行的子任务序列，并动态调整计划以应对执行中的意外，直接决定了系统的实用性。

本文基于小玉米在实际工程中的实践经验，深入剖析 AI Agent Task Decomposition（任务分解）与 Planning（规划）系统的完整技术栈。

一、为什么需要显式规划系统？

1.1 无规划 Agent 的局限性

直接让 LLM 执行复杂多步任务而不经过规划阶段，会出现：

1.2 规划带来的核心收益

二、任务分解核心范式

2.1 三层任务分解架构

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional, Dict, Any, Callable
from abc import ABC, abstractmethod
import json
from datetime import datetime


class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"


class DependencyType(Enum):
    SEQUENTIAL = "sequential"      # B 需要 A 的结果
    PARALLEL = "parallel"          # A 和 B 无依赖
    CONDITIONAL = "conditional"    # 根据 A 的结果决定是否执行 B


@dataclass
class Task:
    """原子任务单元"""
    id: str
    description: str
    agent_prompt: str
    dependencies: List[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    input_data: Optional[Dict] = None
    output_data: Optional[Dict] = None
    expected_tools: List[str] = field(default_factory=list)
    max_retries: int = 3
    timeout_seconds: int = 120
    priority: int = 0  # 数字越小优先级越高


@dataclass
class TaskPlan:
    """完整的任务计划"""
    goal: str
    tasks: List[Task]
    dependency_graph: Dict[str, List[str]]
    context: Dict[str, Any] = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.now)
    current_task_id: Optional[str] = None

2.2 三种分解策略

策略一：LLM 驱动的自然语言分解

最通用的方法，适合开放域任务：

class LLMDecomposer:
    """使用 LLM 将高层目标分解为子任务"""
    
    DECOMPOSITION_PROMPT = """你是一个任务分解专家。将以下用户目标分解为可执行的子任务序列。

用户目标: {goal}

可用工具: {tools}

请以 JSON 格式返回任务列表:
{{
  "tasks": [
    {{
      "id": "task_1",
      "description": "任务描述",
      "prompt": "执行该任务的详细指导",
      "expected_tools": ["tool_name"],
      "dependencies": []
    }}
  ],
  "reasoning": "分解思路"
}}

要求：
1. 每个任务必须是原子性的（一个 LLM 调用可完成）
2. 明确标注任务间的依赖关系
3. 优先并行执行无依赖的任务
4. 总任务数不超过 8 个"""
    
    def __init__(self, llm_call_fn: Callable):
        self.llm = llm_call_fn
    
    def decompose(self, goal: str, tools: List[str]) -> TaskPlan:
        prompt = self.DECOMPOSITION_PROMPT.format(
            goal=goal, tools=", ".join(tools)
        )
        response = self.llm(prompt)
        parsed = json.loads(response)
        
        tasks = []
        deps = {}
        for t_data in parsed["tasks"]:
            task = Task(
                id=t_data["id"],
                description=t_data["description"],
                agent_prompt=t_data["prompt"],
                dependencies=t_data.get("dependencies", []),
                expected_tools=t_data.get("expected_tools", []),
            )
            tasks.append(task)
            deps[task.id] = task.dependencies
        
        return TaskPlan(
            goal=goal,
            tasks=tasks,
            dependency_graph=deps,
        )

策略二：基于模板的结构化分解

适合重复性高、流程固定的任务：

class TemplateDecomposer:
    """基于预定义模板的任务分解"""
    
    TEMPLATES = {
        "financial_analysis": {
            "description": "金融数据分析",
            "steps": [
                {"id": "fetch_data", "description": "获取原始数据"},
                {"id": "clean_data", "description": "数据清洗与预处理",
                 "depends_on": ["fetch_data"]},
                {"id": "calculate_metrics", "description": "计算核心财务指标",
                 "depends_on": ["clean_data"]},
                {"id": "trend_analysis", "description": "趋势分析",
                 "depends_on": ["calculate_metrics"]},
                {"id": "generate_report", "description": "生成分析报告",
                 "depends_on": ["trend_analysis"]},
            ]
        },
        "code_review": {
            "description": "代码审查",
            "steps": [
                {"id": "static_analysis", "description": "静态代码分析"},
                {"id": "security_scan", "description": "安全漏洞扫描"},
                {"id": "logic_review", "description": "业务逻辑审查",
                 "depends_on": ["static_analysis"]},
                {"id": "architecture_review", "description": "架构评审",
                 "depends_on": ["logic_review", "security_scan"]},
                {"id": "summary_report", "description": "审查总结",
                 "depends_on": ["architecture_review"]},
            ]
        },
        "research_paper": {
            "description": "学术论文分析",
            "steps": [
                {"id": "abstract_analysis", "description": "摘要理解"},
                {"id": "methodology_review", "description": "方法论分析"},
                {"id": "result_analysis", "description": "结果数据分析"},
                {"id": "critical_eval", "description": "批判性评估",
                 "depends_on": ["methodology_review", "result_analysis"]},
                {"id": "summary", "description": "综合总结",
                 "depends_on": ["critical_eval"]},
            ]
        }
    }
    
    def decompose(self, goal: str, tools: List[str]) -> Optional[TaskPlan]:
        goal_lower = goal.lower()
        template_key = None
        
        for key, template in self.TEMPLATES.items():
            if (template["description"] in goal or 
                key.replace("_", " ") in goal):
                template_key = key
                break
        
        if not template_key:
            return None
        
        template = self.TEMPLATES[template_key]
        tasks = []
        deps = {}
        for step in template["steps"]:
            dep_list = step.get("depends_on", [])
            task = Task(
                id=step["id"],
                description=step["description"],
                agent_prompt=(
                    f"基于以下目标执行{step['description']}: {goal}"
                ),
                dependencies=dep_list,
            )
            tasks.append(task)
            deps[task.id] = dep_list
        
        return TaskPlan(goal=goal, tasks=tasks, dependency_graph=deps)

策略三：混合分解器（推荐方案）

结合 LLM 的灵活性和模板的效率：

class HybridDecomposer:
    """混合任务分解器 - 先用模板匹配，失败则回退到 LLM"""
    
    def __init__(self, llm_decomposer: LLMDecomposer,
                 template_decomposer: TemplateDecomposer):
        self.llm = llm_decomposer
        self.template = template_decomposer
    
    def decompose(self, goal: str, tools: List[str]) -> TaskPlan:
        # 第一步：尝试模板匹配
        plan = self.template.decompose(goal, tools)
        if plan:
            return plan
        # 第二步：回退到 LLM 分解
        return self.llm.decompose(goal, tools)

三、依赖图与执行调度

3.1 依赖图构建

from collections import deque


class DependencyGraph:
    """任务依赖关系图 - 支持 DAG 检测和拓扑排序"""
    
    def __init__(self, tasks: List[Task],
                 dependencies: Dict[str, List[str]]):
        self.tasks = {t.id: t for t in tasks}
        self.dependencies = dependencies
    
    def get_ready_tasks(self, completed_ids: set) -> List[Task]:
        """获取所有依赖已满足的可执行任务"""
        ready = []
        for task in self.tasks.values():
            if task.id in completed_ids:
                continue
            if task.status != TaskStatus.PENDING:
                continue
            deps = self.dependencies.get(task.id, [])
            if all(d in completed_ids for d in deps):
                ready.append(task)
        ready.sort(key=lambda t: t.priority)
        return ready
    
    def has_cycle(self) -> bool:
        """检测是否存在循环依赖"""
        WHITE, GRAY, BLACK = 0, 1, 2
        color = {t.id: WHITE for t in self.tasks.values()}
        
        def dfs(node_id):
            color[node_id] = GRAY
            for dep_id in self.dependencies.get(node_id, []):
                if color.get(dep_id) == GRAY:
                    return True
                if color.get(dep_id) == WHITE:
                    if dfs(dep_id):
                        return True
            color[node_id] = BLACK
            return False
        
        for task_id in self.tasks:
            if color[task_id] == WHITE:
                if dfs(task_id):
                    return True
        return False
    
    def topo_sort(self) -> List[Task]:
        """拓扑排序 - 获取最优执行顺序"""
        if self.has_cycle():
            raise ValueError("任务存在循环依赖，无法排序")
        
        in_degree = {}
        for task in self.tasks.values():
            in_degree[task.id] = 0
        for task_id, deps in self.dependencies.items():
            in_degree[task_id] = len(deps)
        
        queue = deque([t.id for t in self.tasks.values()
                      if in_degree[t.id] == 0])
        result = []
        
        while queue:
            node_id = queue.popleft()
            result.append(self.tasks[node_id])
            for other_id in self.tasks:
                if node_id in self.dependencies.get(other_id, []):
                    in_degree[other_id] -= 1
                    if in_degree[other_id] == 0:
                        queue.append(other_id)
        return result

3.2 并行执行调度器

import asyncio
from concurrent.futures import ThreadPoolExecutor, as_completed


class PlanExecutor:
    """任务计划执行器 - 支持并行调度和动态重规划"""
    
    def __init__(self, max_parallel: int = 3):
        self.max_parallel = max_parallel
        self.results: Dict[str, Any] = {}
        self.failed_tasks: Dict[str, str] = {}
    
    async def execute_plan(self, plan: TaskPlan,
                          task_runner: Callable[[Task, Dict], Any]
                          ) -> Dict[str, Any]:
        """
        执行整个任务计划
        - 自动识别可并行执行的任务
        - 支持任务级重试
        - 遇到不可恢复错误时触发重规划
        """
        dep_graph = DependencyGraph(plan.tasks, plan.dependency_graph)
        
        if dep_graph.has_cycle():
            return {"status": "failed", "error": "循环依赖"}
        
        completed = set()
        loop = asyncio.get_event_loop()
        executor = ThreadPoolExecutor(max_workers=self.max_parallel)
        
        while len(completed) < len(plan.tasks):
            ready_tasks = dep_graph.get_ready_tasks(completed)
            
            if not ready_tasks and len(completed) < len(plan.tasks):
                return {"status": "blocked",
                        "completed": list(completed),
                        "failed": self.failed_tasks}
            
            batch = ready_tasks[:self.max_parallel]
            futures = {}
            
            for task in batch:
                task.status = TaskStatus.IN_PROGRESS
                future = loop.run_in_executor(
                    executor, task_runner, task, self.results
                )
                futures[future] = task
            
            for future in as_completed(futures):
                task = futures[future]
                try:
                    result = future.result()
                    if result and result.get("success"):
                        task.status = TaskStatus.COMPLETED
                        task.output_data = result.get("data")
                        self.results[task.id] = result.get("data")
                        completed.add(task.id)
                    else:
                        error = (
                            result.get("error", "Unknown error")
                            if result else "Timed out"
                        )
                        self._handle_task_failure(task, error)
                        if task.max_retries <= 0:
                            completed.add(task.id)
                except Exception as e:
                    self._handle_task_failure(task, str(e))
                    if task.max_retries <= 0:
                        completed.add(task.id)
        
        return {
            "status": "completed",
            "results": self.results,
            "failed": self.failed_tasks,
            "execution_time": datetime.now().isoformat(),
        }
    
    def _handle_task_failure(self, task: Task, error: str):
        task.max_retries -= 1
        if task.max_retries > 0:
            task.status = TaskStatus.PENDING
        else:
            task.status = TaskStatus.FAILED
            self.failed_tasks[task.id] = error

四、自适应重规划系统

4.1 触发重规划的条件

@dataclass
class ReplanTrigger:
    """重规划触发条件"""
    task_failed_too_many: bool = False
    context_changed: bool = False
    unexpected_result: bool = False
    user_intervention: bool = False
    cost_overrun: bool = False


class ReplanningEngine:
    """自适应重规划引擎"""
    
    def __init__(self, decomposer: HybridDecomposer):
        self.decomposer = decomposer
    
    def should_replan(self, trigger: ReplanTrigger,
                     plan: TaskPlan, results: Dict) -> bool:
        reasons = []
        if trigger.task_failed_too_many:
            reasons.append("critical_task_failure")
        if trigger.context_changed:
            reasons.append("context_shift")
        if trigger.unexpected_result:
            for task_id, data in results.items():
                if (data and "confidence" in data
                    and data["confidence"] < 0.3):
                    reasons.append(f"low_confidence_{task_id}")
        if trigger.cost_overrun:
            total_tokens = sum(
                r.get("tokens", 0) for r in results.values() if r
            )
            if total_tokens > plan.context.get("token_budget", 100000):
                reasons.append("cost_overrun")
        return len(reasons) > 0
    
    def replan(self, plan: TaskPlan, results: Dict[str, Any],
              failed: Dict[str, str]) -> TaskPlan:
        """
        基于已执行的结果重新规划剩余任务
        策略: 保留已完成的任务结果，只对未完成/失败的部分重新分解
        """
        completed_ids = set(results.keys())
        remaining = [
            t for t in plan.tasks
            if t.id not in completed_ids and t.id not in failed
        ]
        
        if not remaining:
            return plan
        
        remaining_context = {
            "completed": results,
            "failed": failed,
            "original_goal": plan.goal,
            "remaining_tasks": [t.description for t in remaining],
        }
        
        new_goal = (
            f"原始目标: {plan.goal}\n"
            f"已完成: {json.dumps(
                {k: v.get('summary', '')
                 for k, v in results.items() if v})}\n"
            f"待完成: {json.dumps(
                [t.description for t in remaining])}\n"
            f"失败任务: {json.dumps(failed)}\n"
            "请根据已完成的结果，重新规划剩余任务的执行方案。"
        )
        
        new_plan = self.decomposer.decompose(
            new_goal,
            list(set(t for t in plan.tasks for t in t.expected_tools))
        )
        return new_plan

4.2 完整规划-执行-重规划 Pipeline

class PlanningPipeline:
    """生产级 Planning Pipeline"""
    
    def __init__(self, decomposer: HybridDecomposer,
                 executor: PlanExecutor,
                 replanner: ReplanningEngine,
                 max_replan_rounds: int = 3):
        self.decomposer = decomposer
        self.executor = executor
        self.replanner = replanner
        self.max_replan_rounds = max_replan_rounds
        self.metrics = {
            "total_planning_time": 0,
            "replan_count": 0,
            "task_count": 0,
            "failed_task_count": 0,
        }
    
    async def run(self, goal: str, tools: List[str],
                 task_runner: Callable,
                 token_budget: int = 100000) -> Dict[str, Any]:
        """
        完整的规划执行流程
        1. 分解任务 → 2. 执行计划 → 3. 评估结果
        → 4. 如有必要重规划 → 回到 2 → 5. 返回最终结果
        """
        start_time = datetime.now()
        
        plan = self.decomposer.decompose(goal, tools)
        plan.context["token_budget"] = token_budget
        self.metrics["task_count"] = len(plan.tasks)
        
        for round_idx in range(1, self.max_replan_rounds + 1):
            print(f"[Pipeline] 轮次 {round_idx}/{self.max_replan_rounds}")
            print(f"  任务数量: {len(plan.tasks)}")
            
            result = await self.executor.execute_plan(plan, task_runner)
            self.metrics["failed_task_count"] = len(
                result.get("failed", {})
            )
            
            trigger = ReplanTrigger(
                task_failed_too_many=len(
                    result.get("failed", {})
                ) > 0,
                context_changed=self._context_has_changed(
                    plan, result
                ),
                unexpected_result=self._has_unexpected_results(result),
            )
            
            if self.replanner.should_replan(
                trigger, plan, result.get("results", {})
            ):
                self.metrics["replan_count"] += 1
                plan = self.replanner.replan(
                    plan,
                    result.get("results", {}),
                    result.get("failed", {})
                )
            else:
                break
        
        elapsed = (datetime.now() - start_time).total_seconds()
        self.metrics["total_planning_time"] = elapsed
        
        return {
            "status": "completed",
            "final_results": self.executor.results,
            "failed_tasks": self.executor.failed_tasks,
            "metrics": self.metrics,
            "plan": plan,
        }

五、性能基准测试

5.1 三种分解策略对比

5.2 有无规划系统的性能对比

5.3 重规划轮次对成功率的影响

结论：3 轮重规划是性价比最优配置，再增加会因过度重规划导致边际收益递减。

六、生产级最佳实践

6.1 架构设计原则

1. Plan → Execute → Verify → Replan：将规划和执行分离为独立阶段
2. Task Atomicity：每个任务限制为单次 LLM 调用或单次工具调用
3. Dependency as DAG：所有依赖关系建模为有向无环图
4. Graceful Degradation：部分任务失败时不影响已完成的任务
5. Observability First：每个规划、执行、重规划事件都记录可观测数据

6.2 常见陷阱与解决方案

6.3 监控与可观测性

@dataclass
class PlanningMetrics:
    """规划系统的可观测数据"""
    planning_rounds: int
    total_tasks: int
    parallel_tasks: int
    sequential_tasks: int
    replan_count: int
    avg_task_duration: float
    total_duration: float
    success_rate: float
    token_consumption: int
    critical_path_tasks: List[str]

七、未来趋势

Task Decomposition & Planning 在 2026-2027 年将迎来以下关键演进：

1. MCTS 集成规划：将蒙特卡洛树搜索引入 Agent 规划空间，在决策时进行多步前瞻
2. 分层强化学习规划：通过 HRL（Hierarchical RL）自动学习任务分解策略，无需人工设计模板
3. 多 Agent 协作规划：多个 Specialist Agent 各自规划后合并，通过 Debate 机制达成共识
4. 持续规划（Continuous Planning）：规划不再是一次性的前置步骤，而是与执行并行的持续过程
5. 成本感知规划：规划器自动在任务质量、Token 成本和执行延迟之间做 Pareto 最优权衡
6. 代码即规划：将规划结果直接编译为可执行的 DAG 代码，消除规划-执行的语义鸿沟

本文基于小玉米在实际 AI Agent 工程中的实践经验总结，代码示例均为生产级可用的实现模式。

发布时间：2026-06-28 | 作者：🌽 小玉米