6.2 规划能力：任务分解与步骤编排

原文：https://www.mstx.cn/harness/436.html · 创建：2026-09-08 · 更新：2026-09-08 · 阅读：1074

ReAct的局限：走一步看一步

在6.1节中，你实现了ReAct模式——Agent在每一步先思考再行动，根据工具结果调整下一步。这种模式对于线性推进的任务很有效，但它有一个明显的局限：Agent缺乏全局视野。

考虑这个任务："帮我准备下周的产品评审会，需要收集竞品数据、整理用户反馈、准备演示文稿。"

用ReAct模式处理，Agent会：

思考：需要收集竞品数据 → 调用工具查竞品
观察：拿到了数据
思考：接下来整理用户反馈 → 调用工具查反馈
观察：拿到了反馈
思考：接下来准备演示文稿 → 调用工具生成文稿
完成

这个流程本身没问题。但问题在于：Agent在每一步只看到了眼前的子任务，没有在开始时形成一个完整的执行计划。

这导致几个潜在问题：

遗漏步骤：Agent可能在执行中忘记某个子任务（比如忘了整理用户反馈就直接准备文稿）
顺序不当：Agent可能先准备文稿再收集数据，导致文稿需要重做
无法预估工作量：用户在开始时不知道Agent需要多少步才能完成
难以跟踪进度：用户不知道Agent进行到哪一步了，还剩多少

规划能力解决的核心问题：在执行前先想清楚做什么，再按计划推进。

规划的本质：从"反应式"到"计划式"

ReAct（反应式）：        Planner（计划式）：

思考→行动→观察          规划→执行→验证
  ↑         ↓             ↑       ↓
  └───循环───┘             └──按计划推进──┘

每一步基于上一步的结果    每一步基于预先制定的计划
适合不确定性高的任务      适合步骤相对明确的任务

这两种模式不是互斥的。最有效的Agent通常结合两者：先做一个粗略的计划，在执行中根据实际情况调整计划。

实现规划：让LLM先制定计划

第一步：添加规划提示词

PLANNER_SYSTEM_PROMPT = """
你是一个具备规划能力的智能助手。在开始执行复杂任务前，请先制定计划。

计划格式：
[计划]
1. 步骤1的描述
2. 步骤2的描述
3. 步骤3的描述
...

[开始执行]
然后逐步执行计划中的每个步骤。

规则：
1. 如果任务简单（2步以内完成），不需要制定计划，直接执行
2. 如果任务复杂（3步以上），先制定计划，再执行
3. 计划应该具体、可执行，避免模糊的表述
4. 执行中如果发现计划需要调整，可以修改计划并说明原因
5. 完成所有步骤后，总结执行结果
"""

第二步：实现计划提取

python

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

@dataclass
class PlanStep:
    """计划中的一个步骤"""
    step_number: int
    description: str
    status: str = "pending"  # pending / in_progress / completed / skipped
    result: Optional[str] = None

@dataclass
class Plan:
    """一个完整的执行计划"""
    steps: List[PlanStep]
    created_at: str = field(default_factory=lambda: __import__('datetime').datetime.now().isoformat())

    def current_step(self) -> Optional[PlanStep]:
        """获取当前正在执行的步骤"""
        for step in self.steps:
            if step.status == "in_progress":
                return step
        return None

    def next_pending_step(self) -> Optional[PlanStep]:
        """获取下一个待执行的步骤"""
        for step in self.steps:
            if step.status == "pending":
                return step
        return None

    def mark_completed(self, step_number: int, result: str = None):
        """标记步骤为已完成"""
        for step in self.steps:
            if step.step_number == step_number:
                step.status = "completed"
                step.result = result

    def mark_in_progress(self, step_number: int):
        """标记步骤为执行中"""
        for step in self.steps:
            if step.step_number == step_number:
                step.status = "in_progress"

    def all_completed(self) -> bool:
        """检查是否所有步骤都已完成"""
        return all(step.status == "completed" for step in self.steps)

    def progress(self) -> str:
        """返回进度描述"""
        completed = sum(1 for s in self.steps if s.status == "completed")
        return f"{completed}/{len(self.steps)}"

class PlanExtractor:
    """从LLM输出中提取计划"""

    def extract(self, text: str) -> Optional[Plan]:
        """从文本中提取计划"""
        if "[计划]" not in text:
            return None

        # 提取[计划]到[开始执行]之间的内容
        pattern = r'\\[计划\\](.*?)(?:\\[开始执行\\]|$)'
        match = re.search(pattern, text, re.DOTALL)
        if not match:
            return None

        plan_text = match.group(1)
        steps = []

        # 按行解析步骤
        for line in plan_text.strip().split('\n'):
            line = line.strip()
            if not line:
                continue

            # 匹配 "1. xxx" 或 "1) xxx" 或 "1、xxx" 格式
            step_match = re.match(r'^(\d+)[.、)]\s*(.+)', line)
            if step_match:
                step_num = int(step_match.group(1))
                step_desc = step_match.group(2).strip()
                steps.append(PlanStep(
                    step_number=step_num,
                    description=step_desc,
                ))

        if not steps:
            return None

        return Plan(steps=steps)

第三步：实现规划执行的Runtime

python

class PlannerRuntime:
    """带规划能力的Agent运行时"""

    def __init__(self, llm_client, tool_registry, memory,
                 max_iterations: int = 15,
                 verbose: bool = True):
        self.client = llm_client
        self.tools = tool_registry
        self.memory = memory
        self.max_iterations = max_iterations
        self.verbose = verbose

        self.current_plan: Optional[Plan] = None
        self.plan_extractor = PlanExtractor()
        self.execution_log = []

    def run(self, user_input: str) -> str:
        """执行带规划的任务"""
        self.memory.initialize(user_input)

        # 第1轮：让LLM制定计划
        plan_response = self._call_llm(force_planning=True)
        plan_text = plan_response.choices[0].message.content or ""

        self.current_plan = self.plan_extractor.extract(plan_text)

        if self.current_plan:
            if self.verbose:
                print(f"\n制定计划（{len(self.current_plan.steps)}步）：")
                for step in self.current_plan.steps:
                    print(f"  {step.step_number}. {step.description}")

            # 将计划加入记忆，作为执行指导
            plan_summary = self._format_plan_for_context()
            self.memory.add_system_message(plan_summary)
        else:
            if self.verbose:
                print("\n任务简单，无需制定计划，直接执行。")

        # 后续轮次：执行计划
        return self._execute_loop()

    def _execute_loop(self) -> str:
        """执行计划的主循环"""
        for iteration in range(self.max_iterations):
            if self.verbose:
                print(f"\n--- 执行第{iteration + 1}轮 ---")

            # 更新计划状态
            self._update_plan_status()

            # 将当前计划状态注入上下文
            if self.current_plan:
                status_text = self._format_plan_status()
                self.memory.add_system_message(status_text)

            # 调用LLM
            response = self._call_llm()
            message = response.choices[0].message

            if message.tool_calls:
                self.memory.add_llm_message(message)

                for tool_call in message.tool_calls:
                    name = tool_call.function.name
                    args = json.loads(tool_call.function.arguments)

                    if self.verbose:
                        print(f"  执行：{name}({args})")

                    result = self.tools.execute(name, args)

                    # 根据工具执行结果更新计划
                    self._update_plan_from_tool_result(name, result.content)

                    self.memory.add_tool_result(result)
            else:
                final_answer = message.content or ""
                if self.verbose:
                    print(f"  回答：{final_answer}")
                return final_answer

        return "达到最大迭代次数。"

    def _call_llm(self, force_planning: bool = False):
        """调用LLM"""
        messages = self.memory.get_context()

        if force_planning:
            # 在提示词中强制要求规划
            planning_instruction = (
                "\n\n在开始执行前，请先分析任务的复杂度。"
                "如果任务需要3步以上完成，请按以下格式制定计划：\n"
                "[计划]\n1. 步骤1\n2. 步骤2\n...\n[开始执行]\n"
                "如果任务简单，直接回答。"
            )
            messages = messages + [{"role": "system", "content": planning_instruction}]

        return self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            tools=self.tools.get_schemas(),
            temperature=0.0,
        )

    def _format_plan_for_context(self) -> str:
        """将计划格式化为上下文文本"""
        if not self.current_plan:
            return ""

        lines = ["[执行计划]"]
        for step in self.current_plan.steps:
            lines.append(f"{step.step_number}. {step.description}")
        lines.append("[按计划逐步执行]")
        return "\n".join(lines)

    def _format_plan_status(self) -> str:
        """格式化当前计划状态"""
        if not self.current_plan:
            return ""

        lines = [f"[计划进度：{self.current_plan.progress()}"]
        for step in self.current_plan.steps:
            status_mark = {
                "pending": "⬜",
                "in_progress": "▶️",
                "completed": "✅",
                "skipped": "⏭️",
            }.get(step.status, "❓")
            lines.append(f"{status_mark} 步骤{step.step_number}: {step.description}")
        lines.append("]")
        return "\n".join(lines)

    def _update_plan_status(self):
        """更新计划状态"""
        if not self.current_plan:
            return

        # 找到第一个pending的步骤，标记为in_progress
        for step in self.current_plan.steps:
            if step.status == "pending":
                step.status = "in_progress"
                break

    def _update_plan_from_tool_result(self, tool_name: str, result: str):
        """根据工具执行结果更新计划"""
        if not self.current_plan:
            return

        # 简单策略：如果工具执行成功，标记当前步骤为完成
        current = self.current_plan.current_step()
        if current and "失败" not in result and "错误" not in result:
            current.status = "completed"
            current.result = result[:100]

    def get_execution_report(self) -> Dict[str, Any]:
        """生成执行报告"""
        if not self.current_plan:
            return {"planned": False}

        return {
            "planned": True,
            "total_steps": len(self.current_plan.steps),
            "completed_steps": sum(1 for s in self.current_plan.steps if s.status == "completed"),
            "step_details": [
                {
                    "number": s.step_number,
                    "description": s.description,
                    "status": s.status,
                }
                for s in self.current_plan.steps
            ],
        }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173

运行示例

用户：帮我准备下周的产品评审会，需要收集竞品数据、整理用户反馈、准备演示文稿

制定计划（3步）：
  1. 收集竞品数据
  2. 整理用户反馈
  3. 准备演示文稿

--- 执行第1轮 ---
[计划进度：0/3]
⬜ 步骤1: 收集竞品数据
⬜ 步骤2: 整理用户反馈
⬜ 步骤3: 准备演示文稿

  执行：search_competitive_data({"keyword": "智能手表"})
  观察：找到5个竞品的相关信息...

--- 执行第2轮 ---
[计划进度：1/3]
✅ 步骤1: 收集竞品数据
▶️ 步骤2: 整理用户反馈
⬜ 步骤3: 准备演示文稿

  执行：collect_user_feedback({"product": "智能手表"})
  观察：收集到23条用户反馈...

--- 执行第3轮 ---
[计划进度：2/3]
✅ 步骤1: 收集竞品数据
✅ 步骤2: 整理用户反馈
▶️ 步骤3: 准备演示文稿

  执行：generate_presentation({"content": "竞品数据+用户反馈"})
  观察：演示文稿已生成...

--- 执行第4轮 ---
[计划进度：3/3]
✅ 步骤1: 收集竞品数据
✅ 步骤2: 整理用户反馈
✅ 步骤3: 准备演示文稿

回答：已完成所有准备工作。演示文稿已生成，包含竞品分析和用户反馈汇总。

规划与ReAct的结合

规划不是替代ReAct，而是增强它。在实际执行中，每个计划步骤的内部仍然使用ReAct循环——Agent在步骤内"思考-行动-观察"，但整体上遵循预先制定的计划。

┌─────────────────────────────────────────┐
│ 整体规划（Plan）                         │
│ 步骤1 → 步骤2 → 步骤3                    │
│                                         │
│ 每个步骤内部（ReAct）：                   │
│   步骤1：思考→行动→观察→思考→行动→完成   │
│   步骤2：思考→行动→观察→完成             │
│   步骤3：思考→行动→观察→思考→完成       │
└─────────────────────────────────────────┘

这种双层结构是当前Agent工程中最有效的执行模式：外层规划提供全局视野和进度跟踪，内层ReAct提供灵活的问题解决能力。

计划调整：执行中发现问题怎么办

好的规划不是僵化的。Agent应该在执行中发现计划需要调整时主动修改计划：

python

def _update_plan_from_tool_result(self, tool_name: str, result: str):
    """根据工具执行结果更新计划"""
    if not self.current_plan:
        return

    current = self.current_plan.current_step()
    if current and "失败" not in result and "错误" not in result:
        current.status = "completed"
        current.result = result[:100]

    # 检查结果中是否暗示需要调整计划
    adjustment_signals = [
        ("数据不足", "需要增加数据收集步骤"),
        ("无法获取", "需要寻找替代数据源"),
        ("超出预期", "需要增加分析步骤"),
    ]

    for signal, new_step_desc in adjustment_signals:
        if signal in result:
            # 添加新步骤
            new_step = PlanStep(
                step_number=len(self.current_plan.steps) + 1,
                description=new_step_desc,
            )
            self.current_plan.steps.append(new_step)
            if self.verbose:
                print(f"  计划调整：{signal}，新增步骤：{new_step_desc}")
            break

何时使用规划

规划不是万能的。以下场景适合使用规划：

任务步骤相对明确：用户的需求可以分解为清晰的子任务
任务需要多步完成：步骤超过3步，有遗漏风险
需要进度可见性：用户想知道Agent进行到哪一步了
任务有依赖关系：某些步骤必须在其他步骤之前完成

以下场景不适合使用规划：

高度不确定的任务：Agent无法在开始时预知所有步骤（此时用ReAct更灵活）
简单直接的任务：一步或两步就能完成（规划反而增加开销）
探索性任务：Agent需要在执行中根据发现决定方向

经验法则：当用户的任务描述包含"准备""整理""分析"等需要多步完成的动词时，规划是合适的。当用户的任务是简单的查询或计算时，跳过规划。

本节小结

规划能力让Agent从"反应式执行"进化为"计划式执行"。本节实现了：

计划提取：从LLM输出中解析结构化的执行计划
计划跟踪：实时更新每个步骤的状态（待执行/执行中/已完成）
进度注入：将计划状态注入上下文，让LLM始终知道进度
计划调整：根据执行中的发现动态调整计划

核心原则：规划的价值不在于"制定一个完美的计划"，而在于让执行过程有结构、可跟踪、可审计。即使计划在执行中需要调整，它也比没有计划提供了更好的可控性。

在下一节中，我们将讨论反思机制——如何让Agent在执行后评估自己的表现，从错误中学习，并在必要时修正方向。

ReAct的局限：走一步看一步 ​

规划的本质：从"反应式"到"计划式" ​

实现规划：让LLM先制定计划 ​

规划与ReAct的结合 ​

计划调整：执行中发现问题怎么办 ​

何时使用规划 ​

本节小结 ​

ReAct的局限：走一步看一步

规划的本质：从"反应式"到"计划式"

实现规划：让LLM先制定计划

规划与ReAct的结合

计划调整：执行中发现问题怎么办

何时使用规划

本节小结