在上一篇《Plan-and-Execute Agent》中,我们已经实现了:

Planner
↓
生成 Task List
↓
Executor
↓
逐步执行

这已经比传统 ReAct 更强了。

因为:

Agent 开始具备“先规划,再执行”的能力

但这里仍然有一个非常大的问题:


静态 Planning 的致命缺陷

很多初学者以为:

只要生成 Plan
然后按顺序执行
就是 Planning Agent

实际上:

真正的生产级 Planning Agent
一定是“动态规划(Dynamic Planning)”

原因很简单:

现实世界是变化的。


一个真实问题

比如用户:

帮我分析 OpenAI、Anthropic、Google 最近 AI Agent 产品布局,
并输出一份对比报告

Planner 初始生成:

1. 搜索 OpenAI Agent 产品
2. 搜索 Anthropic Agent 产品
3. 搜索 Google Agent 产品
4. 汇总对比

看起来没问题。

但执行时:


执行到第2步

Executor 发现:

Anthropic 最近发布了 MCP Protocol

这是非常关键的新信息。

于是:

原计划已经不够了

需要新增:

- 调研 MCP Protocol
- 分析 MCP 对 Agent 的意义
- 对比 OpenAI Function Calling

这时候:

Agent 必须动态修改计划

否则:

最终结果一定不完整

这就是动态规划(Re-Planning)

真正的 Agent:

不是:

先想完所有事情
再机械执行

而是:

边执行
边观察
边修正计划

这其实非常像人类。


ReAct 与 Planning 的真正关系

很多教程会把:

ReAct
Plan-and-Execute

当成对立方案。

实际上:

生产环境中,
两者通常是融合的。

真正高级 Agent 往往是:

Planning 负责全局任务拆解

ReAct 负责局部动态决策

也就是:

宏观 Planning
+
微观 ReAct

这是现代 Agent Engineering 的核心模式。


本篇你将学会

这一篇我们会实现:

一个真正“动态规划”的 Agent

包括:


核心能力

1. 初始 Planning

Planner 生成任务列表


2. Executor 执行任务

逐个执行 Task


3. 动态 Re-Planning

执行过程中:

发现新信息
→ 修改 Plan
→ 插入新任务

4. State 实时更新

State 中维护:

plan
current_task
completed_tasks
observations

5. ReAct 混合模式

Executor 内部:

Thought
Action
Observation

动态决定是否继续调用 Tool。


最终架构(非常重要)

我们的目标架构:

                ┌──────────────┐
                │   Planner    │
                └──────┬───────┘
                       │
                       ▼
                ┌──────────────┐
                │   Executor   │
                └──────┬───────┘
                       │
            ┌──────────┴──────────┐
            ▼                     ▼
     是否需要重规划?         当前任务完成?
            │                     │
            ▼                     ▼
      Re-Planning           下一个任务
            │
            ▼
      更新 Plan

注意:

这里已经不是:

固定流程

而是:

动态工作流(Dynamic Workflow)

这才是 LangGraph 真正强大的地方。


为什么 LangGraph 特别适合 Planning Agent?

因为:

Planning Agent 本质就是:
State Machine + Workflow Engine

而 LangGraph:

恰好就是:

Graph-Based State Machine

这也是为什么:


AgentExecutor 不适合复杂 Planning

传统:

AgentExecutor

更适合:

单轮 ReAct

但:

复杂 Planning
动态状态控制
条件分支
Re-Planning

这些:

AgentExecutor 很难优雅实现

LangGraph 则天然适合。


一、定义 State(核心)

先定义动态 Planning 的状态。


为什么 State 特别重要?

因为:

Planning Agent = 长生命周期 Agent

状态会越来越复杂。

所以:

State 设计
决定 Agent 上限

生产级 State 设计

from typing import TypedDict, List, Optional
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
from typing_extensions import Annotated


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

    user_input: str

    # 当前完整计划
    plan: List[str]

    # 当前执行任务
    current_task: Optional[str]

    # 已完成任务
    completed_tasks: List[str]

    # 执行观察结果
    observations: List[str]

    # 是否需要重规划
    need_replan: bool

为什么这样设计?

这是非常典型的:

Planner / Executor 解耦


plan

保存完整任务队列:

[
    "搜索 OpenAI Agent",
    "搜索 Anthropic MCP",
    "对比分析"
]

current_task

Executor 当前执行:

"搜索 Anthropic MCP"

completed_tasks

避免重复执行:

[
    "搜索 OpenAI Agent"
]

observations

保存执行过程中发现的新信息:

[
    "Anthropic 发布 MCP"
]

这些观察:

会影响未来规划。


need_replan

动态控制:

是否重新规划

这是:

Dynamic Workflow 的核心。


二、Planner Node

先实现 Planner。


Planner 的职责

Planner 不是执行任务。

Planner 负责:

任务拆解(Task Decomposition)

例如:

用户:

分析 AI Agent 框架

Planner:

1. 调研 LangGraph
2. 调研 AutoGen
3. 调研 CrewAI
4. 输出对比

Planner Prompt

PLANNER_PROMPT = """
你是一个专业 AI Planning Agent。

请把用户目标拆解为多个可执行任务。

要求:

1. 任务必须具体
2. 任务必须可执行
3. 按执行顺序输出
4. 使用 JSON List 返回

用户需求:

{input}
"""

Planner Node 实现

def planner_node(state: AgentState):

    user_input = state["user_input"]

    prompt = PLANNER_PROMPT.format(
        input=user_input
    )

    response = llm.invoke(prompt)

    tasks = json.loads(response.content)

    return {
        "plan": tasks,
        "current_task": tasks[0] if tasks else None
    }

工程问题:LLM 输出不稳定

很多人这里会踩坑:

json.loads(response.content)

直接炸掉。

因为:

LLM 可能输出:

```json
[
  ...
]

或者:

```text
这里是你的任务:
[
 ...
]

生产环境必须:

使用 Structured Output

例如:

llm.with_structured_output()

或者:

PydanticOutputParser

不要直接裸解析 JSON。

这是生产级 Agent 的基本要求。


三、Executor Node

接下来是核心。


Executor 才是真正的 Agent

很多人误以为:

Planner 最重要

实际上:

Executor 才是 Agent 的灵魂

因为:

真正与环境交互的是 Executor

Executor 的职责

负责:

1. 执行当前任务
2. 调用 Tool
3. 分析结果
4. 判断是否重规划
5. 更新状态

这其实已经开始接近:

ReAct Agent

了。


Executor Prompt

EXECUTOR_PROMPT = """
你是一个 AI 执行 Agent。

当前任务:

{task}

已完成任务:

{completed_tasks}

执行观察:

{observations}

请执行当前任务。
"""

Executor 实现

def executor_node(state: AgentState):

    current_task = state["current_task"]

    prompt = EXECUTOR_PROMPT.format(
        task=current_task,
        completed_tasks=state["completed_tasks"],
        observations=state["observations"]
    )

    result = llm.invoke(prompt)

    observation = result.content

    completed = state["completed_tasks"]
    completed.append(current_task)

    return {
        "observations": state["observations"] + [observation],
        "completed_tasks": completed
    }

这里还不是真正生产级

因为:

现在 Executor:

还只是 LLM

下一步:

我们会升级:

Executor + Tool Calling + ReAct

这才是真正动态 Agent。


四、动态 Re-Planning(核心)

终于来到最关键部分。


为什么需要 Re-Planning?

因为:

执行过程中:

可能发现:
- 新信息
- 错误路径
- 更优方案
- 缺失步骤

所以:

Plan 必须可修改

Re-Planning Prompt

REPLAN_PROMPT = """
你是一个动态规划 Agent。

当前计划:

{plan}

已完成任务:

{completed}

观察结果:

{observations}

请判断:

1. 是否需要修改计划
2. 是否需要新增任务
3. 是否需要删除任务

返回:

{
    "need_replan": true,
    "updated_plan": [...]
}
"""

Re-Planning Node

def replan_node(state: AgentState):

    prompt = REPLAN_PROMPT.format(
        plan=state["plan"],
        completed=state["completed_tasks"],
        observations=state["observations"]
    )

    result = llm.invoke(prompt)

    data = json.loads(result.content)

    return {
        "plan": data["updated_plan"],
        "need_replan": data["need_replan"]
    }

这就是动态 Agent 的本质

注意:

这里已经开始出现:

Agent 自我修正

能力。

这与:

传统 Workflow

有本质区别。


五、Graph 编排(重点)

现在开始进入:

LangGraph 真正强大的部分


Graph 流程

START
  ↓
Planner
  ↓
Executor
  ↓
RePlan Check
  ├── 继续执行
  ├── 重新规划
  └── 结束

Conditional Edge

这是关键。


为什么 Conditional Edge 极其重要?

因为:

Planning Agent:

本质不是固定流程

而是:

动态状态驱动

路由函数

def should_replan(state: AgentState):

    if state["need_replan"]:
        return "replan"

    if len(state["plan"]) > len(state["completed_tasks"]):
        return "executor"

    return END

Graph 编写

from langgraph.graph import StateGraph, START, END

builder = StateGraph(AgentState)

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

builder.add_edge(START, "planner")

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

builder.add_conditional_edges(
    "executor",
    should_replan,
    {
        "replan": "replan",
        "executor": "executor",
        END: END
    }
)

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

graph = builder.compile()

注意这里非常关键

我们已经实现:

动态循环 Workflow

这与:

普通 Chain

完全不同。


六、升级:Executor + ReAct 混合模式

现在:

我们的 Executor:

还只是“调用一次 LLM”

生产环境不够。

真正做法:


Executor 内部嵌套 ReAct

结构:

Planning(宏观)
    ↓
ReAct(微观)

这是现代 Agent 的主流架构。


为什么这样设计?

因为:

Planner:

擅长任务拆解

但:

不擅长实时决策

ReAct:

擅长:
- Tool Calling
- 动态观察
- 环境反馈

所以:

两者组合效果最好。


混合架构

Planner
  ↓
Task

Executor(ReAct)
  ↓
Thought
Action
Observation

Executor 内部

可以直接:

react_agent.invoke()

例如:

def executor_node(state: AgentState):

    current_task = state["current_task"]

    result = react_agent.invoke({
        "messages": [
            HumanMessage(content=current_task)
        ]
    })

    return {
        "observations": [
            result["messages"][-1].content
        ]
    }

这就是现代生产级 Agent

你会发现:

整个系统开始变成:

Graph
    ↓
Node
    ↓
Agent
        ↓
Tool

这已经是真正:

Agent System

而不是:

Prompt Demo

了。


七、生产环境中的核心问题

接下来讲最重要的:

工程问题


1. Re-Planning 死循环

最常见问题:

Agent 一直重规划

永不结束。


典型原因

Planner 每次:

都新增任务

导致:

任务永远做不完

生产级方案

必须增加:

最大规划次数

replan_count
max_replan

例如:

if state["replan_count"] > 5:
    return END

2. Task Explosion(任务爆炸)

另一个经典问题:

任务越拆越多

最后:

几百个 Task

Token 爆炸。


解决方案

必须:

限制:

- 最大 Task 数
- 最大层级
- 最大递归深度

3. Context 污染

动态 Planning 最大隐患:

Observation 越来越长

导致:

Prompt 爆炸

正确做法

必须:

Observation Summary

定期:

压缩历史观察

类似:

Conversation Summary Memory

4. Planner 与 Executor 冲突

经典问题:

Planner:

搜索 Google

Executor:

发现 Bing 更好

怎么办?


生产级做法

必须定义:

Agent Authority

例如:


方案1:Executor 可修改 Plan

更灵活。


方案2:只能 Planner 修改

更稳定。


八、为什么动态 Planning 是未来核心

因为:

未来 Agent:

一定会越来越复杂。

例如:


Deep Research

OpenAI Deep Research:

本质就是:

动态 Planning Agent

Manus

Manus 的核心:

也是:

长链路动态执行

Claude Computer Use

也是:

观察
→ 调整
→ 重规划

本质已经不是 Prompt

而是:

AI Workflow Engine


九、本篇核心总结

这一篇非常关键。

因为:

你已经开始真正进入:

Agent Engineering 核心领域


你学会了什么?


1. 动态 Planning

不是固定 Task List。

而是:

执行中实时修正

2. Re-Planning

Agent:

观察环境
→ 修改计划

3. LangGraph 的真正价值

LangGraph:

不是:

Chain 工具

而是:

Workflow State Machine


4. ReAct + Planning 混合模式

现代 Agent 主流架构:

Planning(宏观)
+
ReAct(微观)

5. 生产级问题

你已经开始接触:

- 死循环
- Task Explosion
- Context 污染
- 状态控制

这些:

才是真正的 Agent 工程问题。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐