任务规划
约 1826 字大约 6 分钟
Agent虾学智能体入门
2026-03-08
任务规划(Task Decomposition)
本系列第五篇,深入讲解智能体如何分解复杂任务并制定执行计划。
为什么需要任务规划?
复杂任务无法一步完成,需要分解为多个子任务并有序执行。任务规划是智能体的核心能力之一。
示例:预订旅行
用户:帮我预订下周五去上海的旅行
复杂任务分解:
1. 确定日期(下周五是哪天?)
2. 查询机票
3. 查询酒店
4. 比较选项
5. 确认预订
6. 发送确认信息任务分解方法
1. 链式分解(Chain Decomposition)
将任务分解为线性步骤,按顺序执行。
from typing import List, Callable
from dataclasses import dataclass
@dataclass
class Task:
"""任务定义"""
id: str
description: str
action: Callable
dependencies: List[str] = None
status: str = "pending"
class ChainPlanner:
"""链式任务规划器"""
def __init__(self, llm):
self.llm = llm
self.tasks = []
def plan(self, goal: str) -> List[Task]:
"""将目标分解为任务链"""
prompt = f"""
将以下目标分解为具体步骤,以 JSON 数组格式返回:
目标:{goal}
格式:
[
{{"id": "1", "description": "步骤描述"}},
{{"id": "2", "description": "步骤描述"}}
]
"""
response = self.llm.generate(prompt)
steps = eval(response) # 实际应用中使用 json.loads
tasks = []
for i, step in enumerate(steps):
task = Task(
id=step["id"],
description=step["description"],
action=self._get_action(step["description"]),
dependencies=[steps[i-1]["id"]] if i > 0 else None
)
tasks.append(task)
self.tasks = tasks
return tasks
def _get_action(self, description: str) -> Callable:
"""根据描述获取对应行动"""
# 简化实现,实际应用中可能需要更复杂的映射
return lambda: f"执行: {description}"
def execute(self) -> List[dict]:
"""按顺序执行任务"""
results = []
for task in self.tasks:
print(f"[执行] Task {task.id}: {task.description}")
result = task.action()
task.status = "completed"
results.append({
"task_id": task.id,
"result": result
})
return results
# 使用示例
# planner = ChainPlanner(llm=openai_client)
# tasks = planner.plan("帮我预订下周五去上海的旅行")
# results = planner.execute()2. 树状分解(Tree Decomposition)
将任务分解为树状结构,支持并行执行。
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class TreeNode:
"""任务树节点"""
id: str
description: str
children: List['TreeNode']
status: str = "pending"
result: Optional[str] = None
class TreePlanner:
"""树状任务规划器"""
def __init__(self, llm):
self.llm = llm
def plan(self, goal: str) -> TreeNode:
"""将目标分解为任务树"""
prompt = f"""
将以下目标分解为任务树,支持并行执行的任务放在同一层级:
目标:{goal}
返回 JSON 格式:
{{
"id": "root",
"description": "主目标",
"children": [
{{
"id": "1",
"description": "子任务1",
"children": []
}},
{{
"id": "2",
"description": "子任务2(可与1并行)",
"children": []
}}
]
}}
"""
response = self.llm.generate(prompt)
tree_data = eval(response)
return self._build_tree(tree_data)
def _build_tree(self, data: dict) -> TreeNode:
"""递归构建任务树"""
return TreeNode(
id=data["id"],
description=data["description"],
children=[self._build_tree(child) for child in data.get("children", [])]
)
def execute_parallel(self, node: TreeNode, executor) -> str:
"""并行执行任务树"""
# 如果有子任务,先执行子任务
if node.children:
import asyncio
async def execute_children():
tasks = [
asyncio.to_thread(self.execute_parallel, child, executor)
for child in node.children
]
results = await asyncio.gather(*tasks)
return results
results = asyncio.run(execute_children())
node.result = f"子任务完成: {len(results)} 个"
else:
# 叶子节点:执行具体任务
node.result = executor(node.description)
node.status = "completed"
return node.result
# 使用示例
# planner = TreePlanner(llm)
# tree = planner.plan("准备一次产品发布会")
# result = planner.execute_parallel(tree, executor=lambda x: f"完成: {x}")3. DAG 分解(有向无环图)
最灵活的分解方式,支持复杂的依赖关系。
from collections import defaultdict, deque
class DAGPlanner:
"""DAG 任务规划器"""
def __init__(self, llm):
self.llm = llm
self.graph = defaultdict(list)
self.in_degree = defaultdict(int)
self.tasks = {}
def plan(self, goal: str) -> dict:
"""将目标分解为 DAG"""
prompt = f"""
将以下目标分解为任务图,标明依赖关系:
目标:{goal}
返回 JSON 格式:
{{
"tasks": [
{{"id": "A", "description": "任务A"}},
{{"id": "B", "description": "任务B"}},
{{"id": "C", "description": "任务C", "depends_on": ["A", "B"]}}
]
}}
"""
response = self.llm.generate(prompt)
plan_data = eval(response)
# 构建图
for task in plan_data["tasks"]:
self.tasks[task["id"]] = task["description"]
depends_on = task.get("depends_on", [])
for dep in depends_on:
self.graph[dep].append(task["id"])
self.in_degree[task["id"]] += 1
if task["id"] not in self.in_degree:
self.in_degree[task["id"]] = 0
return {
"tasks": self.tasks,
"graph": dict(self.graph),
"in_degree": dict(self.in_degree)
}
def execute(self, executor) -> List[str]:
"""拓扑排序执行(支持并行)"""
results = []
queue = deque([
task_id for task_id, degree in self.in_degree.items()
if degree == 0
])
while queue:
# 获取当前可执行的任务(入度为0)
current = queue.popleft()
print(f"[执行] Task {current}: {self.tasks[current]}")
result = executor(self.tasks[current])
results.append(result)
# 更新依赖此任务的其他任务
for next_task in self.graph[current]:
self.in_degree[next_task] -= 1
if self.in_degree[next_task] == 0:
queue.append(next_task)
return results
# 使用示例
# planner = DAGPlanner(llm)
# plan = planner.plan("开发一个Web应用")
# results = planner.execute(executor=lambda x: f"完成: {x}")规划策略
1. Plan-and-Solve
先制定完整计划,再逐步执行。
class PlanAndSolve:
"""Plan-and-Solve 策略"""
def __init__(self, llm):
self.llm = llm
def run(self, goal: str) -> str:
# 1. 规划阶段
plan = self._create_plan(goal)
print(f"[计划]\n{plan}")
# 2. 执行阶段
results = []
for step in plan.split("\n"):
if step.strip():
result = self._execute_step(step)
results.append(result)
return "\n".join(results)
def _create_plan(self, goal: str) -> str:
"""制定计划"""
prompt = f"""
为以下目标制定详细计划:
目标:{goal}
要求:
1. 分解为具体步骤
2. 每步可执行
3. 步骤之间有逻辑顺序
计划:
"""
return self.llm.generate(prompt)
def _execute_step(self, step: str) -> str:
"""执行单个步骤"""
prompt = f"执行以下步骤:{step}"
return self.llm.generate(prompt)2. ReAct Planning
推理与行动交织,动态调整计划。
class ReActPlanner:
"""ReAct 动态规划"""
def __init__(self, llm, tools):
self.llm = llm
self.tools = tools
def run(self, goal: str, max_steps: int = 10):
"""动态规划和执行"""
context = f"目标:{goal}\n"
for i in range(max_steps):
# 思考下一步
thought = self._think(context)
print(f"[思考] {thought}")
# 判断是否完成
if "任务完成" in thought or "目标达成" in thought:
return self._summarize(context)
# 选择行动
action, action_input = self._decide_action(thought)
if action == "finish":
return action_input
# 执行行动
observation = self._execute(action, action_input)
print(f"[观察] {observation}")
# 更新上下文
context += f"\n步骤{i+1}: {thought}\n行动: {action}\n结果: {observation}"
return "达到最大步骤数"
def _think(self, context: str) -> str:
"""思考下一步"""
prompt = f"{context}\n\n下一步应该做什么?"
return self.llm.generate(prompt)
def _decide_action(self, thought: str):
"""决定行动"""
# 解析思考和行动
return "search", {"query": thought}
def _execute(self, action: str, action_input: dict) -> str:
"""执行行动"""
if action in self.tools:
return str(self.tools[action](**action_input))
return "未知行动"
def _summarize(self, context: str) -> str:
"""总结结果"""
return f"任务完成:{context[:200]}..."常见规划框架
LangChain Plan-and-Execute
from langchain_experimental.plan_and_execute import (
PlanAndExecute,
load_agent_executor,
load_chat_planner
)
from langchain_openai import ChatOpenAI
# 创建规划器
llm = ChatOpenAI(model="gpt-4")
planner = load_chat_planner(llm)
executor = load_agent_executor(llm, tools, verbose=True)
# 创建 Plan-and-Execute Agent
agent = PlanAndExecute(planner=planner, executor=executor, verbose=True)
# 运行
agent.run("帮我研究 AI Agent 的最新进展并写一份报告")BabyAGI
class BabyAGI:
"""简化版 BabyAGI"""
def __init__(self, llm, task_executor):
self.llm = llm
self.executor = task_executor
self.task_list = []
self.memory = []
def run(self, objective: str, max_iterations: int = 5):
"""运行 BabyAGI 循环"""
# 初始任务
self.task_list = [{"id": 1, "task": f"开始执行目标:{objective}"}]
for i in range(max_iterations):
if not self.task_list:
break
# 1. 获取下一个任务
task = self.task_list.pop(0)
print(f"[执行任务] {task['task']}")
# 2. 执行任务
result = self.executor(task["task"])
self.memory.append({"task": task, "result": result})
# 3. 存储结果
print(f"[结果] {result[:100]}...")
# 4. 生成新任务
new_tasks = self._generate_tasks(objective, result)
self.task_list.extend(new_tasks)
# 5. 优先级排序
self._prioritize_tasks(objective)
return self._summarize()
def _generate_tasks(self, objective: str, last_result: str) -> list:
"""基于结果生成新任务"""
prompt = f"""
目标:{objective}
已完成:{last_result}
生成3个后续任务(JSON数组):
[{{"id": 1, "task": "任务描述"}}]
"""
response = self.llm.generate(prompt)
try:
return eval(response)
except:
return []
def _prioritize_tasks(self, objective: str):
"""对任务优先级排序"""
# 简化:按 ID 排序
self.task_list.sort(key=lambda x: x.get("id", 0))
def _summarize(self) -> str:
"""总结执行结果"""
return f"完成 {len(self.memory)} 个任务"小结
- 任务规划是智能体处理复杂任务的核心能力
- 三种分解方式:链式(顺序)、树状(并行)、DAG(复杂依赖)
- 两种策略:Plan-and-Solve(先规划再执行)、ReAct(边想边做)
下一篇
记忆系统 - 学习智能体的短期/长期记忆机制