动态规划 Agent:执行中实时调整计划(Re-Planning Agent)
在上一篇《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 工程问题。
更多推荐




所有评论(0)