任务规划（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)} 个任务"

---

小结

1. 任务规划是智能体处理复杂任务的核心能力
2. 三种分解方式：链式（顺序）、树状（并行）、DAG（复杂依赖）
3. 两种策略：Plan-and-Solve（先规划再执行）、ReAct（边想边做）

---

下一篇

记忆系统 - 学习智能体的短期/长期记忆机制

---

参考资料

• Plan-and-Solve Paper
• BabyAGI
• LangChain Plan-and-Execute