一、为什么 Agent 需要“任务分解”？

前面我们已经实现了：

• ReAct Agent
• Tool Calling
• LangGraph Workflow
• 手写 ReAct Graph

但是你会发现一个非常严重的问题：

---

ReAct 的核心短板

ReAct 非常适合：

短链路任务

例如：

• 查询天气
• 调用搜索
• 简单计算
• 一次性 Tool 调用

因为它的模式本质上是：

Thought
→ Action
→ Observation
→ 下一轮决策

它是一种：

边想边做

的机制。

这在简单任务里非常有效。

但问题是：

---

二、复杂任务下，ReAct 会开始失控

例如用户问：

请帮我研究一下：
2026 年 AI Agent 行业的发展趋势，
并重点分析：

1. OpenAI
2. Anthropic
3. Google
4. 国内 Agent 创业公司

最后生成一份总结报告。

你会发现：

普通 ReAct Agent 很容易出现：

---

典型问题1：任务遗漏

Agent 可能：

• 只研究了 OpenAI
• 忘记 Anthropic
• 没做最终总结

因为：

它没有全局任务视图

---

典型问题2：上下文混乱

随着 Tool 调用越来越多：

messages 会越来越长

导致：

• 推理质量下降
• Tool 选择错误
• 开始跑偏

---

典型问题3：无法控制执行顺序

例如：

应该先调研
再总结
最后输出

但 ReAct 可能：

研究到一半直接输出结论

因为：

ReAct 没有“计划层”

---

三、Planning Agent 的核心思想

Planning Agent 的核心升级：

先规划
再执行

也就是：

Plan-and-Execute

架构会变成：

用户请求
    ↓
Planner
    ↓
生成任务列表
    ↓
Executor
    ↓
逐个执行任务
    ↓
汇总结果

这其实非常像：

项目经理 + 执行团队

---

四、任务分解（Task Decomposition）是什么？

所谓：

任务分解

本质上就是：

把复杂任务拆成多个可执行的小任务

例如：

用户请求：

分析 AI Agent 行业

Planner 会自动拆成：

1. 调研 OpenAI Agent 战略
2. 调研 Anthropic Agent 战略
3. 调研 Google Agent 战略
4. 调研中国 Agent 创业公司
5. 汇总行业趋势
6. 生成最终报告

这就是：

Task List

也是 Planning Agent 的核心。

---

五、为什么“任务分解”是 Agent 工程化核心？

这是很多人忽略的一点。

真正的大型 Agent 系统：

核心不是 Prompt
而是 Workflow Control

任务分解的本质其实是：

Workflow Planning

也就是：

工作流规划

这是整个 Agent Engineering 的核心能力。

---

六、任务分解的三种常见模式

在生产环境里，常见有三种：

---

1. 静态任务分解

例如：

[
    "搜索资料",
    "总结内容",
    "生成报告"
]

特点：

• 固定流程
• 最稳定
• 最容易调试

缺点：

不够智能

---

2. LLM 动态任务分解（最常见）

让模型自己生成：

task list

例如：

[
    "研究 OpenAI",
    "研究 Anthropic",
    "总结行业趋势"
]

这是目前主流方案。

因为：

灵活性最高

---

3. 动态重规划（Re-Planning）

执行过程中：

发现任务不够

于是：

重新生成任务

例如：

执行到一半发现：
缺少中国市场数据

于是 Planner 自动新增：

调研中国 AI Agent 创业公司

这是高级 Planning Agent 的核心能力。

后面我们会专门讲。

---

七、任务分解 Prompt 设计（非常重要）

很多人做 Planning Agent 最大的问题：

Planner 输出不稳定

例如：

今天输出：

1. 搜索
2. 总结

明天输出：

先看看情况再说

原因很简单：

Prompt 不够工程化

---

八、工程级 Planner Prompt

下面是一个生产环境里更稳定的 Planner Prompt。

---

Planner Prompt

PLANNER_PROMPT = """
你是一个专业的任务规划 Agent。

你的职责：

1. 将复杂任务拆解成多个子任务
2. 每个子任务必须：
   - 清晰
   - 可执行
   - 原子化
3. 子任务之间应该有合理顺序
4. 不要输出解释
5. 只输出 JSON

输出格式：

[
  {
    "id": 1,
    "task": "搜索 OpenAI Agent 战略"
  },
  {
    "id": 2,
    "task": "总结 OpenAI Agent 产品方向"
  }
]
"""

---

九、为什么“只输出 JSON”非常重要？

因为：

Planning 的本质是 Workflow Control

不是聊天。

如果 Planner 输出：

好的，我来帮你分析。

1. OpenAI
2. Anthropic

你会发现：

json.loads()

直接炸掉。

所以：

Planner 必须结构化输出

这是生产环境的关键。

---

十、Task State 设计（核心）

接下来进入：

LangGraph 工程化核心

我们需要设计：

Planning State

---

十一、Planning Agent 的 State 结构

from typing import TypedDict, List


class AgentState(TypedDict):
    user_input: str

    tasks: List[str]

    current_task: str

    completed_tasks: List[str]

    final_answer: str

---

十二、为什么 Planning Agent 必须强化 State？

因为：

Planning Agent 已经不是“聊天机器人”了

而是：

工作流系统

所以：

State = Workflow Runtime

这是整个 LangGraph 的核心思想。

---

十三、Task Queue 思想（极其重要）

注意这里：

tasks: List[str]

其实本质上就是：

任务队列（Task Queue）

例如：

[
    "研究 OpenAI",
    "研究 Anthropic",
    "生成总结"
]

Executor 每次：

取一个任务
→ 执行
→ 标记完成
→ 继续下一个

这就是：

Queue-based Agent Architecture

这是很多高级 Agent 系统的核心。

---

十四、第一个任务分解 Agent（完整代码）

下面开始实现：

Planner + Executor

---

安装依赖

pip install langgraph langchain openai

---

完整代码

import json

from typing import TypedDict, List

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph
from langgraph.graph import START, END


llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0
)


PLANNER_PROMPT = """
你是一个专业任务规划助手。

请将用户任务拆解成多个子任务。

要求：
1. 子任务必须清晰
2. 子任务必须可执行
3. 按执行顺序排列
4. 只输出 JSON

输出格式：
[
  {
    "task": "任务1"
  }
]
"""


class AgentState(TypedDict):
    user_input: str

    tasks: List[str]

    current_task: str

    completed_tasks: List[str]

    final_answer: str


# =========================
# Planner Node
# =========================


def planner_node(state: AgentState):
    user_input = state["user_input"]

    prompt = f"""
{PLANNER_PROMPT}

用户任务：
{user_input}
"""

    response = llm.invoke(prompt)

    tasks_json = json.loads(response.content)

    tasks = [item["task"] for item in tasks_json]

    print("\n========== Planner ==========")
    print(tasks)

    return {
        "tasks": tasks,
        "completed_tasks": []
    }


# =========================
# Executor Node
# =========================


def executor_node(state: AgentState):
    tasks = state["tasks"]

    completed_tasks = state["completed_tasks"]

    if not tasks:
        return {
            "final_answer": "所有任务执行完成"
        }

    current_task = tasks[0]

    print("\n========== Executor ==========")
    print(f"当前任务: {current_task}")

    # 模拟执行
    result = f"已完成: {current_task}"

    completed_tasks.append(result)

    remaining_tasks = tasks[1:]

    return {
        "tasks": remaining_tasks,
        "current_task": current_task,
        "completed_tasks": completed_tasks
    }


# =========================
# Router
# =========================


def route_after_executor(state: AgentState):
    if len(state["tasks"]) == 0:
        return "end"

    return "continue"


# =========================
# Build Graph
# =========================

builder = StateGraph(AgentState)

builder.add_node("planner", planner_node)
builder.add_node("executor", executor_node)

builder.add_edge(START, "planner")

builder.add_edge("planner", "executor")

builder.add_conditional_edges(
    "executor",
    route_after_executor,
    {
        "continue": "executor",
        "end": END
    }
)


graph = builder.compile()


# =========================
# Run
# =========================

result = graph.invoke({
    "user_input": "研究 2026 AI Agent 行业趋势并生成总结"
})


print("\n========== Final ==========")
print(result)

---

十五、这个 Agent 的运行流程

整个执行过程：

START
  ↓
Planner
  ↓
生成 tasks
  ↓
Executor
  ↓
取第一个 task
  ↓
执行
  ↓
更新 tasks
  ↓
继续循环
  ↓
END

这已经不是：

聊天系统

而是：

工作流引擎

了。

---

十六、Graph 架构图

               ┌──────────────┐
               │    START     │
               └──────┬───────┘
                      │
                      ▼
              ┌──────────────┐
              │   Planner    │
              └──────┬───────┘
                     │
                     ▼
              ┌──────────────┐
              │   Executor   │
              └──────┬───────┘
                     │
          ┌──────────┴──────────┐
          │                     │
          ▼                     ▼
   还有任务？                无任务
          │                     │
          ▼                     ▼
      Executor                 END

---

十七、这里最重要的工程思想

很多人看到这里：

会觉得：

不就是循环执行吗？

但实际上：

这里已经进入：

Agent Runtime

领域了。

因为：

Planner 负责决策
Executor 负责执行
Graph 负责调度
State 负责状态管理

这已经是：

完整的 Agent Workflow System

了。

---

十八、复杂研究任务示例（非常典型）

例如用户输入：

请研究：

1. AI Agent 行业趋势
2. OpenAI Agent 产品
3. Anthropic MCP
4. Google Gemini Agent
5. 国内创业公司

最后生成报告

Planner 可能输出：

[
  {
    "task": "调研 AI Agent 行业趋势"
  },
  {
    "task": "分析 OpenAI Agent 产品布局"
  },
  {
    "task": "研究 Anthropic MCP 生态"
  },
  {
    "task": "分析 Google Gemini Agent 能力"
  },
  {
    "task": "调研中国 AI Agent 创业公司"
  },
  {
    "task": "生成最终总结报告"
  }
]

这已经非常接近：

真正的研究型 Agent

了。

---

十九、多步骤数据处理案例

任务分解并不只是“研究任务”。

在企业里还有大量：

数据处理型 Agent

例如：

分析销售数据并生成周报

Planner 可能拆成：

1. 加载销售数据
2. 清洗异常值
3. 统计销售趋势
4. 分析 TOP 商品
5. 生成图表
6. 输出报告

这其实已经非常像：

Airflow / Dagster

这种 Workflow 系统。

所以：

Agent Engineering
≈ AI Workflow Engineering

这是非常重要的认知升级。

---

二十、Planner 最大的坑：任务粒度

这是生产环境里非常容易翻车的地方。

例如：

Planner 输出：

研究 AI

这就太大了。

Executor 根本无法执行。

---

二十一、正确的任务粒度

正确应该是：

1. 搜索 OpenAI Agent 产品
2. 总结 OpenAI Agent 能力
3. 搜索 Anthropic MCP
4. 总结 MCP 特点

也就是：

任务必须原子化

这是 Planner Prompt 的核心。

---

二十二、生产环境里的 Planner 最佳实践

---

1. 强制 JSON 输出

一定要：

结构化输出

否则系统极不稳定。

---

2. 限制任务数量

否则 Planner 会：

无限拆任务

例如：

最多 10 个任务

---

3. 限制任务粒度

避免：

研究 AI

这种超大任务。

---

4. Planner 和 Executor 解耦

这是最关键的一点。

不要：

一个 Node 既规划又执行

否则：

系统会越来越不可控

---

二十三、为什么 LangGraph 特别适合 Planning Agent？

因为：

Planning Agent 本质是状态机

而 LangGraph 最大优势就是：

Graph + State + Conditional Edge

这刚好就是：

Workflow Runtime

的核心。

---

二十四、Planning Agent 与 ReAct 的本质区别

---

ReAct

特点：

边想边做

适合：

• 简单任务
• Tool 调用
• 实时交互

---

Planning Agent

特点：

先规划
再执行

适合：

• 长任务
• 多步骤任务
• 研究任务
• 数据处理
• 企业 Workflow

---

二十五、真正的大型 Agent 都在强化 Planning

2025~2026 的一个非常明显趋势：

Agent 正在从 Chat 走向 Workflow

例如：

• OpenAI Deep Research
• Manus
• Devin
• Claude Research

它们共同特点：

都有 Planning Layer

因为：

没有 Planning
Agent 无法处理长链路复杂任务

---

二十六、下一篇会继续升级什么？

下一篇我们会继续进入：

动态重规划（Re-Planning）

也就是：

执行过程中自动调整任务

例如：

发现信息不足

Agent 自动：

新增任务

这会真正进入：

高级 Planning Agent

领域。

---

二十七、本章总结

这一章非常关键。

因为我们第一次真正进入了：

Workflow-based Agent

阶段。

核心认知升级：

---

1. ReAct 不适合复杂长任务

因为：

没有全局规划能力

---

2. Planning Agent = 先规划再执行

核心：

Planner + Executor

---

3. Task Decomposition 是核心能力

本质：

Workflow Planning

---

4. LangGraph 非常适合 Planning

因为：

Planning 本质就是状态机

---

5. Agent 正在从 ChatBot 走向 Workflow Engine

这是 2026 Agent Engineering 的核心趋势。

---

下一章预告

28. 动态重规划（Re-Planning）

我们会真正开始进入：

高级 Planning Agent

包括：

• 动态新增任务
• Executor 失败重试
• Plan 修正
• Reflection
• Self-Correction
• 自适应 Workflow

这是：

真正高级 Agent 系统

的开始。