任务分解策略
任务分解（Task Decomposition） 是将复杂目标拆分为可执行步骤的关键规划能力。合理的任务分解能让 Agent 高效完成复杂任务，是 Agent 规划模块的核心功能之一。
一、核心概念
1.1 什么是任务分解？
任务分解 是将一个大目标拆分为多个小任务的过程：
┌─────────────────────────────────────────────────────────────┐
│ 任务分解示意 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 输入：大目标 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ "帮我完成一份市场调研报告" │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ 任务分解 │
│ 输出：子任务列表 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 1. 确定调研主题和范围 │ │
│ │ 2. 收集市场数据和竞品信息 │ │
│ │ 3. 分析市场趋势和用户需求 │ │
│ │ 4. 撰写调研报告 │ │
│ │ 5. 审核和修改报告 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1.2 为什么需要任务分解？
| 原因 | 说明 |
|---|---|
| 降低复杂度 | 大任务难以直接执行，小任务更易处理 |
| 明确步骤 | 让执行过程清晰可控 |
| 并行处理 | 独立子任务可以并行执行 |
| 错误隔离 | 单个任务失败不影响整体 |
| 进度可视 | 可以跟踪每个子任务的完成情况 |
1.3 分解粒度的重要性
┌─────────────────────────────────────────────────────────────┐
│ 分解粒度谱系 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 粒度过粗 粒度过细 │
│ ◄──────────────────────────────────────────────────► │
│ │
│ ┌──────────────────────────────────────────────────────┐│
│ │ 过粗示例： ││
│ │ 任务：完成市场调研报告 ││
│ │ 步骤： ││
│ │ 1. 做调研 ││
│ │ 2. 写报告 ││
│ │ ││
│ │ ❌ 步骤太粗，难以执行 ││
│ └──────────────────────────────────────────────────────┘│
│ │
│ ┌──────────────────────────────────────────────────────┐│
│ │ 适中示例： ││
│ │ 任务：完成市场调研报告 ││
│ │ 步骤： ││
│ │ 1. 确定调研主题 ││
│ │ 2. 收集市场数据（搜索工具） ││
│ │ 3. 分析竞品情况（分析工具） ││
│ │ 4. 撰写报告（写作工具） ││
│ │ ││
│ │ ✅ 粒度适中，每步对应工具调用 ││
│ └──────────────────────────────────────────────────────┘│
│ │
│ ┌──────────────────────────────────────────────────────┐│
│ │ 过细示例： ││
│ │ 任务：完成市场调研报告 ││
│ │ 步骤： ││
│ │ 1. 打开浏览器 ││
│ │ 2. 点击搜索框 ││
│ │ 3. 输入关键词 ││
│ │ 4. 点击搜索按钮 ││
│ │ 5. 滚动页面 ││
│ │ ...（几百个步骤） ││
│ │ ││
│ │ ❌ 粒度太细，效率低下 ││
│ └──────────────────────────────────────────────────────┘│
│ │
│ 最佳粒度原则：每个步骤对应一个明确的工具调用 │
│ │
└─────────────────────────────────────────────────────────────┘二、分解策略
2.1 分解策略分类
| 策略 | 描述 | 适用场景 |
|---|---|---|
| 层级分解 | 按抽象层次分解 | 有明确层次结构的任务 |
| 依赖分解 | 按依赖关系分解 | 有明确依赖关系的任务 |
| 能力分解 | 按 Agent 能力边界分解 | 需要特定工具的任务 |
| 时间分解 | 按时间顺序分解 | 有时间依赖的任务 |
2.2 层级分解
┌─────────────────────────────────────────────────────────────┐
│ 层级分解示例 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 目标：开发一个用户管理系统 │
│ │
│ 第一层（目标层）：用户管理系统 │
│ │ │
│ ├─ 第二层（模块层）： │
│ │ ├── 用户注册模块 │
│ │ ├── 用户登录模块 │
│ │ ├── 权限管理模块 │
│ │ └── 数据统计模块 │
│ │ │
│ └─ 第三层（功能层）： │
│ ├── 用户注册模块 │
│ │ ├── 表单验证 │
│ │ ├── 密码加密 │
│ │ └── 数据存储 │
│ ├── 用户登录模块 │
│ │ ├── 身份验证 │
│ │ ├── Token 生成 │
│ │ └── 会话管理 │
│ └── ... │
│ │
│ 优点：结构清晰，易于理解和维护 │
│ 适用：大型系统、复杂项目 │
│ │
└─────────────────────────────────────────────────────────────┘2.3 依赖分解（DAG）
┌─────────────────────────────────────────────────────────────┐
│ 依赖分解示例（DAG） │
├─────────────────────────────────────────────────────────────┤
│ │
│ 任务：准备一场技术分享会议 │
│ │
│ ┌──────────┐ │
│ │ 确定主题 │──────────────────────┐ │
│ └────┬─────┘ │ │
│ │ │ │
│ ↓ ↓ │
│ ┌──────────┐ ┌──────────┐ │
│ │ 准备大纲 │ │ 预定场地 │ │
│ └────┬─────┘ └────┬─────┘ │
│ │ │ │
│ ↓ ↓ │
│ ┌──────────┐ ┌──────────┐ │
│ │ 制作PPT │ │ 发送邀请 │ │
│ └────┬─────┘ └────┬─────┘ │
│ │ │ │
│ └──────────┬─────────────────┘ │
│ ↓ │
│ ┌──────────┐ │
│ │ 进行演讲 │ │
│ └──────────┘ │
│ │
│ DAG 特点： │
│ • 有向无环图，无循环依赖 │
│ • 可以识别可并行执行的任务 │
│ • 支持拓扑排序确定执行顺序 │
│ │
└─────────────────────────────────────────────────────────────┘2.4 能力分解
┌─────────────────────────────────────────────────────────────┐
│ 能力分解示例 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 任务：完成数据分析报告 │
│ │
│ Agent 能力： │
│ ├── 搜索工具：search(query) │
│ ├── 数据处理：process_data(file, operations) │
│ ├── 可视化：create_chart(data, type) │
│ ├── 写作：write_report(content) │
│ └── 发送：send_email(to, content) │
│ │
│ 分解结果： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 1. search("行业数据 2024") → 搜索工具 │ │
│ │ 2. process_data(file, "clean") → 数据处理工具 │ │
│ │ 3. process_data(file, "analyze") → 数据处理工具 │ │
│ │ 4. create_chart(data, "bar") → 可视化工具 │ │
│ │ 5. write_report(analysis) → 写作工具 │ │
│ │ 6. send_email("boss@", report) → 发送工具 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 原则：每个步骤都能被某个现有工具执行 │
│ │
└─────────────────────────────────────────────────────────────┘三、代码实现
3.1 基础任务分解器
"""
任务分解器基础实现
支持多种分解策略
"""
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum
from langchain_openai import ChatOpenAI
class DecompositionStrategy(Enum):
"""分解策略"""
HIERARCHICAL = "hierarchical" # 层级分解
DEPENDENCY = "dependency" # 依赖分解
CAPABILITY = "capability" # 能力分解
@dataclass
class Task:
"""任务节点"""
id: str
name: str
description: str
dependencies: List[str] # 依赖的任务 ID
tool: Optional[str] = None # 使用的工具
status: str = "pending" # pending, running, completed, failed
result: Optional[str] = None
class TaskDecomposer:
"""
任务分解器
将复杂目标分解为可执行的子任务
"""
DECOMPOSITION_PROMPT = """
你是一个任务规划专家。请将以下目标分解为具体的子任务。
目标：{goal}
可用工具：
{tools}
分解要求：
1. 每个子任务应该明确、具体
2. 每个子任务应该能被某个工具执行
3. 子任务之间应该有明确的依赖关系
4. 子任务数量适中，不要过粗或过细
请按以下 JSON 格式输出：
{{
"tasks": [
{{
"id": "task_1",
"name": "任务名称",
"description": "任务描述",
"tool": "工具名称",
"dependencies": []
}},
{{
"id": "task_2",
"name": "任务名称",
"description": "任务描述",
"tool": "工具名称",
"dependencies": ["task_1"]
}}
]
}}
"""
def __init__(
self,
model_name: str = "gpt-4",
strategy: DecompositionStrategy = DecompositionStrategy.HIERARCHICAL
):
"""
初始化任务分解器
Args:
model_name: 模型名称
strategy: 分解策略
"""
self.llm = ChatOpenAI(model=model_name, temperature=0)
self.strategy = strategy
self.available_tools: Dict[str, str] = {}
def register_tool(self, name: str, description: str):
"""注册可用工具"""
self.available_tools[name] = description
def decompose(self, goal: str) -> List[Task]:
"""
分解任务
Args:
goal: 目标描述
Returns:
任务列表
"""
# 构建工具描述
tools_desc = "\n".join([
f"- {name}: {desc}"
for name, desc in self.available_tools.items()
])
# 调用 LLM 进行分解
prompt = self.DECOMPOSITION_PROMPT.format(
goal=goal,
tools=tools_desc if tools_desc else "无限制"
)
response = self.llm.invoke(prompt)
# 解析结果
import json
try:
# 尝试提取 JSON
content = response.content
if "```json" in content:
content = content.split("```json")[1].split("```")[0]
elif "```" in content:
content = content.split("```")[1].split("```")[0]
data = json.loads(content)
tasks = []
for t in data.get("tasks", []):
task = Task(
id=t["id"],
name=t["name"],
description=t["description"],
dependencies=t.get("dependencies", []),
tool=t.get("tool")
)
tasks.append(task)
return tasks
except Exception as e:
# 解析失败，返回默认任务
return [Task(
id="task_1",
name="执行目标",
description=goal,
dependencies=[]
)]
def get_execution_order(self, tasks: List[Task]) -> List[List[Task]]:
"""
获取任务执行顺序（支持并行）
使用拓扑排序，返回每批可并行执行的任务
Returns:
分批的任务列表，每批内的任务可以并行执行
"""
# 构建依赖图
task_map = {t.id: t for t in tasks}
in_degree = {t.id: 0 for t in tasks}
for task in tasks:
for dep in task.dependencies:
if dep in in_degree:
in_degree[task.id] += 1
# 拓扑排序
result = []
remaining = set(in_degree.keys())
while remaining:
# 找出当前入度为 0 的任务
ready = [task_map[tid] for tid in remaining if in_degree[tid] == 0]
if not ready:
# 存在循环依赖
break
result.append(ready)
# 更新依赖
for task in ready:
remaining.remove(task.id)
for other_task in tasks:
if task.id in other_task.dependencies:
in_degree[other_task.id] -= 1
return result
# 使用示例
if __name__ == "__main__":
decomposer = TaskDecomposer()
# 注册工具
decomposer.register_tool("search", "搜索互联网获取信息")
decomposer.register_tool("analyze", "分析数据")
decomposer.register_tool("write", "撰写文档")
decomposer.register_tool("email", "发送邮件")
# 分解任务
tasks = decomposer.decompose(
"完成一份关于 AI 发展趋势的市场调研报告，并发送给老板"
)
print("分解结果：")
for task in tasks:
print(f" [{task.id}] {task.name}")
print(f" 描述：{task.description}")
print(f" 工具：{task.tool}")
print(f" 依赖：{task.dependencies}")
# 获取执行顺序
order = decomposer.get_execution_order(tasks)
print("\n执行顺序（每批可并行）：")
for i, batch in enumerate(order, 1):
print(f" 第{i}批：{[t.name for t in batch]}")3.2 层级分解器
"""
层级分解器实现
支持多级任务分解
"""
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class HierarchicalTask:
"""层级任务节点"""
id: str
name: str
description: str
level: int # 层级深度
parent: Optional[str] = None # 父任务 ID
children: List['HierarchicalTask'] = None
status: str = "pending"
def __post_init__(self):
if self.children is None:
self.children = []
class HierarchicalDecomposer:
"""
层级分解器
将任务按层次结构分解
"""
DECOMPOSITION_PROMPT = """
请将以下任务分解为更具体的子任务。
当前任务：{task}
层级：{level}
分解要求：
1. 生成 2-5 个子任务
2. 子任务应该是当前任务的组成部分
3. 子任务粒度适中
4. 如果任务已经足够简单，返回 "无需分解"
格式：
子任务1: ...
子任务2: ...
（或"无需分解"）
"""
def __init__(
self,
model_name: str = "gpt-4",
max_depth: int = 3,
min_task_size: int = 10 # 最小任务描述长度
):
self.llm = ChatOpenAI(model=model_name, temperature=0)
self.max_depth = max_depth
self.min_task_size = min_task_size
def decompose_recursive(
self,
task: HierarchicalTask,
current_depth: int = 0
) -> HierarchicalTask:
"""递归分解任务"""
# 检查深度限制
if current_depth >= self.max_depth:
return task
# 检查是否需要分解
if len(task.description) < self.min_task_size:
return task
# 调用 LLM 分解
prompt = self.DECOMPOSITION_PROMPT.format(
task=task.description,
level=current_depth
)
response = self.llm.invoke(prompt)
content = response.content
# 检查是否无需分解
if "无需分解" in content or "不需要分解" in content:
return task
# 解析子任务
children = []
for i, line in enumerate(content.split('\n')):
line = line.strip()
if line.startswith('子任务') or line.startswith(f'{i+1}.'):
# 提取任务描述
if ':' in line:
desc = line.split(':', 1)[1].strip()
elif '.' in line:
desc = line.split('.', 1)[1].strip()
else:
continue
if desc:
child = HierarchicalTask(
id=f"{task.id}_{i+1}",
name=f"子任务{i+1}",
description=desc,
level=current_depth + 1,
parent=task.id
)
# 递归分解
child = self.decompose_recursive(child, current_depth + 1)
children.append(child)
task.children = children
return task
def decompose(self, goal: str) -> HierarchicalTask:
"""
分解目标
Args:
goal: 目标描述
Returns:
层级任务树
"""
root = HierarchicalTask(
id="root",
name="主任务",
description=goal,
level=0
)
return self.decompose_recursive(root)
def flatten(self, task: HierarchicalTask) -> List[HierarchicalTask]:
"""将层级任务展平为列表"""
result = [task]
for child in task.children:
result.extend(self.flatten(child))
return result
def print_tree(self, task: HierarchicalTask, indent: int = 0):
"""打印任务树"""
prefix = " " * indent
print(f"{prefix}├─ [{task.id}] {task.name}: {task.description[:50]}...")
for child in task.children:
self.print_tree(child, indent + 1)
# 使用示例
if __name__ == "__main__":
decomposer = HierarchicalDecomposer(max_depth=2)
task_tree = decomposer.decompose(
"开发一个在线商城系统"
)
print("任务树：")
decomposer.print_tree(task_tree)四、粒度控制
4.1 粒度评估标准
| 标准 | 描述 | 示例 |
|---|---|---|
| 可执行性 | 任务能否被现有工具执行 | ✅ 调用搜索工具 / ❌ "思考一下" |
| 独立性 | 任务是否相对独立 | ✅ 单一功能 / ❌ 耦合多个操作 |
| 可验证性 | 任务完成是否可验证 | ✅ 产生文件 / ❌ 不明确输出 |
| 时长适中 | 任务执行时间是否合理 | ✅ 1-10分钟 / ❌ 几小时或几秒 |
4.2 粒度调整策略
┌─────────────────────────────────────────────────────────────┐
│ 粒度调整策略 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 粒度过粗的信号： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ • 任务描述模糊，如"处理数据" │ │
│ │ • 无法映射到具体工具 │ │
│ │ • 执行时需要再次分解 │ │
│ │ • 多个子步骤混在一起 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ 调细 │
│ │ │
│ 粒度过细的信号： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ • 任务描述过于具体，如"点击按钮" │ │
│ │ • 多个任务可以合并为一个工具调用 │ │
│ │ • 任务执行时间极短 │ │
│ │ • 任务数量过多，管理开销大 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ 调粗 │
│ │ │
│ 最佳粒度： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ • 每个任务对应一个工具调用 │ │
│ │ • 任务有明确的输入输出 │ │
│ │ • 任务执行时间适中（分钟级） │ │
│ │ • 任务数量可控（一般 < 20个） │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘4.3 粒度调整代码
class GranularityController:
"""粒度控制器"""
def __init__(
self,
min_tasks: int = 3,
max_tasks: int = 20,
min_description_length: int = 10,
max_description_length: int = 200
):
self.min_tasks = min_tasks
self.max_tasks = max_tasks
self.min_description_length = min_description_length
self.max_description_length = max_description_length
def evaluate_granularity(self, tasks: List[Task]) -> str:
"""评估任务粒度"""
if len(tasks) < self.min_tasks:
return "too_coarse" # 太粗
if len(tasks) > self.max_tasks:
return "too_fine" # 太细
avg_length = sum(len(t.description) for t in tasks) / len(tasks)
if avg_length < self.min_description_length:
return "too_fine" # 描述太短，太细
if avg_length > self.max_description_length:
return "too_coarse" # 描述太长，太粗
return "appropriate" # 适中
def adjust(
self,
tasks: List[Task],
granularity: str
) -> List[Task]:
"""调整任务粒度"""
if granularity == "appropriate":
return tasks
if granularity == "too_coarse":
# 需要细化
return self._refine(tasks)
else:
# 需要合并
return self._merge(tasks)
def _refine(self, tasks: List[Task]) -> List[Task]:
"""细化任务"""
refined = []
for task in tasks:
# 如果任务太粗，尝试分解
if len(task.description) > self.max_description_length:
# 这里可以调用分解器进一步分解
refined.append(task)
else:
refined.append(task)
return refined
def _merge(self, tasks: List[Task]) -> List[Task]:
"""合并任务"""
# 合并相邻的小任务
merged = []
current_batch = []
for task in tasks:
if len(task.description) < self.min_description_length:
current_batch.append(task)
else:
if current_batch:
merged_task = self._merge_batch(current_batch)
merged.append(merged_task)
current_batch = []
merged.append(task)
if current_batch:
merged_task = self._merge_batch(current_batch)
merged.append(merged_task)
return merged
def _merge_batch(self, tasks: List[Task]) -> Task:
"""合并一批任务"""
return Task(
id="_".join(t.id for t in tasks),
name="合并任务",
description="; ".join(t.description for t in tasks),
dependencies=tasks[0].dependencies
)五、常见问题与解决方案
5.1 循环依赖
┌─────────────────────────────────────────────────────────────┐
│ 循环依赖问题 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 问题示例： │
│ Task A 依赖 Task B │
│ Task B 依赖 Task C │
│ Task C 依赖 Task A ← 形成循环 │
│ │
│ 解决方案： │
│ 1. 检测：使用拓扑排序检测循环 │
│ 2. 打破：识别循环链，重新定义依赖 │
│ 3. 合并：将循环中的任务合并为一个 │
│ │
│ 代码检测： │
│ def has_cycle(tasks): │
│ # 使用 DFS 检测环 │
│ visited = set() │
│ rec_stack = set() │
│ │
│ def dfs(task_id): │
│ visited.add(task_id) │
│ rec_stack.add(task_id) │
│ │
│ for dep in get_dependencies(task_id): │
│ if dep not in visited: │
│ if dfs(dep): │
│ return True │
│ elif dep in rec_stack: │
│ return True # 发现环 │
│ │
│ rec_stack.remove(task_id) │
│ return False │
│ │
│ return any(dfs(t.id) for t in tasks if t.id not in visited) │
│ │
└─────────────────────────────────────────────────────────────┘5.2 任务无法映射到工具
┌─────────────────────────────────────────────────────────────┐
│ 工具映射失败 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 问题：分解的任务没有对应的工具可以执行 │
│ │
│ 示例： │
│ 任务："思考用户需求" │
│ 可用工具：search, write, email │
│ → 无对应工具 │
│ │
│ 解决方案： │
│ 1. 重新分解：将抽象任务转为具体操作 │
│ "思考用户需求" → "搜索用户需求分析案例" │
│ 2. 扩展工具：添加新工具支持 │
│ 3. 人工介入：标记为需要人工处理的任务 │
│ │
└─────────────────────────────────────────────────────────────┘六、面试高频问题
Q1: 如何判断任务分解的粒度是否合适？
答案要点：
- 每个任务能否被现有工具执行
- 任务是否有明确的输入输出
- 任务数量是否在可控范围内（一般 < 20）
- 任务执行时间是否适中（分钟级）
Q2: 如何处理任务之间的循环依赖？
答案要点：
- 使用拓扑排序或 DFS 检测循环
- 重新定义依赖关系，打破循环
- 将循环中的任务合并为一个任务
- 使用迭代执行方式处理
Q3: 任务分解和执行如何协作？
答案要点：
- 先分解后执行（Plan-and-Execute）
- 边分解边执行（ReAct）
- 分层分解执行（Hierarchical）
- 根据任务复杂度选择策略
Q4: 如何支持任务的并行执行？
答案要点：
- 使用 DAG 表示依赖关系
- 拓扑排序确定执行顺序
- 同一层级无依赖的任务可并行
- 使用并行执行框架（如 asyncio）
Q5: 任务分解失败的常见原因？
答案要点：
- 目标描述不清晰
- 工具能力不足
- 分解粒度不合适
- 存在循环依赖
- LLM 输出格式错误
七、小结
| 概念 | 一句话总结 | 面试关键词 |
|---|---|---|
| 任务分解 | 将大目标拆分为可执行的小任务 | 规划、粒度控制 |
| 层级分解 | 按抽象层次逐级分解 | 树结构、递归 |
| 依赖分解 | 按 DAG 依赖关系分解 | 拓扑排序、并行执行 |
| 粒度控制 | 保持任务大小适中 | 可执行性、独立性 |
一句话总结：任务分解是 Agent 规划的核心能力，合理的分解粒度能让 Agent 高效完成复杂任务。
最后更新：2026年3月19日