L7 规划设计模式：让 Agent 会"想清楚再动手"

• 难度 ⭐⭐⭐（中级）
• 预计阅读 35 分钟
• 代码量 含三种模式的完整 LangGraph 实现
• 先修 L5（LangGraph）+ L6（多 Agent）

课程目标：掌握三种主流的 Planning 高级模式——Plan-and-Execute、ReWOO、Reflexion，能根据场景选择合适的模式。

一、ReAct 的局限性#

L2 我们学了 ReAct 循环：

Thought → Action → Observation → Thought → Action → Observation → ...

ReAct 的优点：实现简单，循环一次走一步。

ReAct 的缺点：当任务需要多步规划时效率低下。

场景：策划一场海外团建#

用户说："帮我策划明年 Q1 的海外团建，地点泰国，预算 50 万，10 个人，5 天。"

ReAct 怎么做？

Step 1: Thought: 查曼谷到普吉的机票
        Action: search_flight("曼谷-普吉", "2026-01-15")
        Observation: 找到 5 个航班...

Step 2: Thought: 查曼谷的酒店
        Action: search_hotel("曼谷", 5天, 10人)
        Observation: 找到 20 家酒店...

Step 3: Thought: 算一下酒店+机票总价
        Action: calculate(hotel_cost + flight_cost)
        Observation: 总额 38 万...

Step 4: Thought: 还需要查签证和当地交通
        Action: search_visa("泰国")
        Observation: ...

Step 5: Thought: 写一份综合报告
        Action: write_report(...)
        Observation: ...

问题 1：上下文爆炸。每一步的 Observation 都加进 prompt，第 10 步时 prompt 里塞了前面 9 步的所有内容——token 浪费严重。

问题 2：没法提前规划。ReAct 是"走一步看一步"，无法回答"整个计划的依赖关系是什么"。

问题 3：没有反思。如果中间某步算错了（比如机票价格过时），ReAct 会基于错误结果继续走，不会回头检查。

这一课我们学习的三种 Planning 模式正是为了解决这些问题。

二、用一个比喻：三种 Planning 模式 = 三种出行方式#

把"完成复杂任务"比作"从北京去上海"：

三、模式 1：Plan-and-Execute#

核心思想#

先想清楚整个计划，再逐步执行。

把任务拆成两步：

1. Planner（规划器）：一次性生成完整步骤列表（不调工具，纯思考）

2. Executor（执行器）：按步骤依次执行，每步可以调工具

流程图#

用户问题
   ↓
[Planner] → 生成步骤列表 ["步骤1", "步骤2", "步骤3"]
   ↓
[Executor] → 执行步骤1（可能调工具）
   ↓
[Executor] → 执行步骤2
   ↓
[Executor] → 执行步骤3
   ↓
最终答案

完整 LangGraph 实现#

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_tavily import TavilySearch

# ============ 1. 定义 State ============

class PlanState(TypedDict):
    messages: Annotated[list, add_messages]
    plan: list[str]              # Planner 生成的步骤列表
    current_step: int            # 当前执行到第几步
    step_results: list[str]      # 每步的执行结果
    final_answer: str            # 最终汇总

# ============ 2. 准备模型和工具 ============

llm = ChatOpenAI(model="gpt-4o-mini")
search_tool = TavilySearch(max_results=2)
tools = [search_tool]
llm_with_tools = llm.bind_tools(tools)

# ============ 3. Planner 节点 ============

PLANNER_PROMPT = """你是一个 Planner。请基于用户请求生成一个详细的执行计划。

要求：
1. 把任务拆成 3-7 个具体步骤
2. 每步必须是独立的、可执行的
3. 标注每步需要的工具（如 "search_web"、"calculate"）
4. 按依赖顺序排序

输出格式（严格遵守）：
STEP 1: <步骤描述> | TOOL: <工具名>
STEP 2: <步骤描述> | TOOL: <工具名>
...

用户请求：{user_request}
"""

def planner_node(state: PlanState):
    """生成完整计划"""
    user_msg = state["messages"][-1].content
    response = llm.invoke(PLANNER_PROMPT.format(user_request=user_msg))
    
    # 解析输出，提取步骤
    plan = []
    for line in response.content.split("\n"):
        if line.strip().startswith("STEP"):
            plan.append(line.strip())
    
    return {"plan": plan, "current_step": 0, "step_results": []}

# ============ 4. Executor 节点 ============

def executor_node(state: PlanState):
    """执行当前步骤"""
    current = state["current_step"]
    plan = state["plan"]
    
    if current >= len(plan):
        # 所有步骤完成，进入汇总
        return {"current_step": current}
    
    step = plan[current]
    
    EXECUTOR_PROMPT = f"""你是一个 Executor。请执行以下步骤：
    {step}
    
    之前的执行结果：
    {chr(10).join(state['step_results'])}
    
    用合适的工具完成任务。"""
    
    response = llm_with_tools.invoke(EXECUTOR_PROMPT)
    return {
        "messages": [response],
        "step_results": state["step_results"] + [response.content or "（无文本输出）"]
    }

# ============ 5. Tool Node ============

from langgraph.prebuilt import ToolNode
tool_node = ToolNode(tools)

# ============ 6. 条件边 ============

def should_continue_executor(state: PlanState) -> str:
    last_msg = state["messages"][-1] if state["messages"] else None
    if last_msg and last_msg.tool_calls:
        return "tools"
    return "next_step"

def should_continue_plan(state: PlanState) -> str:
    if state["current_step"] >= len(state["plan"]):
        return "summarize"
    return "executor"

# ============ 7. Summarizer 节点 ============

def summarizer_node(state: PlanState):
    """汇总所有步骤结果，生成最终答案"""
    prompt = f"""请基于以下步骤结果生成最终回答：
    
    计划：
    {chr(10).join(state['plan'])}
    
    执行结果：
    {chr(10).join(state['step_results'])}
    
    请用清晰的结构汇总最终答案。"""
    
    response = llm.invoke(prompt)
    return {"final_answer": response.content}

# ============ 8. 装配图 ============

graph_builder = StateGraph(PlanState)

graph_builder.add_node("planner", planner_node)
graph_builder.add_node("executor", executor_node)
graph_builder.add_node("tools", tool_node)
graph_builder.add_node("summarizer", summarizer_node)

graph_builder.add_edge(START, "planner")
graph_builder.add_edge("planner", "executor")
graph_builder.add_conditional_edges(
    "executor",
    should_continue_executor,
    {"tools": "tools", "next_step": "next_step_router"}
)

# 加一个 next_step 节点用于推进步骤计数
def next_step_node(state: PlanState):
    return {"current_step": state["current_step"] + 1}

graph_builder.add_node("next_step_router", next_step_node)
graph_builder.add_conditional_edges(
    "next_step_router",
    should_continue_plan,
    {"executor": "executor", "summarize": "summarizer"}
)

graph_builder.add_edge("tools", "executor")
graph_builder.add_edge("summarizer", END)

graph = graph_builder.compile()

运行示例#

用户问："对比 OpenAI 和 Anthropic 在企业级应用上的定价"

Planner 输出：

STEP 1: 搜索 OpenAI 企业级定价信息 | TOOL: search_web
STEP 2: 搜索 Anthropic 企业级定价信息 | TOOL: search_web
STEP 3: 整理两家公司的定价结构 | TOOL: none
STEP 4: 对比分析，生成最终报告 | TOOL: none

Executor 依次执行，每步可能调工具，最终 Summarizer 汇总。

优缺点#

四、模式 2：ReWOO（Reasoning WithOut Observation）#

核心思想#

把"推理"和"执行"完全分离。

ReWOO 有三个独立模块：

关键洞察#

Planner 看不到工具输出，所以它必须预先假设每步会拿到什么。这是和 ReAct/Plan-and-Execute 的根本区别。

ReWOO vs ReAct：Token 效率对比#

ReAct 的 token 成本（每步重新喂所有历史）：

$$Token_{ReAct} = Q + C + kS + \sum_{j=1}^{k}(k-j)(T_j + A_j + O_j)$$

ReWOO 的 token 成本（推理只跑一次）：

$$Token_{ReWOO} \approx 2Q + 2C + S + \sum_{j=1}^{k}(P_j + E_j)$$

关键差异：ReAct 的 token 增长是二次方（每步都依赖前面所有步骤的 Observation），ReWOO 是线性。

论文实测：在 HotpotQA（多步推理基准）上，ReWOO 比 ReAct 省 5 倍 token，同时准确率高 4%（来源：arXiv:2305.18323）。

流程图#

用户问题
   ↓
[Planner] → 生成带占位的完整计划
   ↓
[Worker] → 并行调用所有工具，填充 #E1, #E2, ...
   ↓
[Solver] → 合并计划和结果，生成答案

完整 LangGraph 实现#

class ReWOOState(TypedDict):
    messages: Annotated[list, add_messages]
    plan: str                # Planner 输出
    evidence: dict[str, str] # Worker 收集的证据
    answer: str              # Solver 输出

PLANNER_PROMPT = """你是一个 Planner。请生成执行计划，每步用 #E1, #E2 占位。
不要假设工具输出，只生成"如果工具返回 X，则下一步是 Y"的形式。

用户问题：{input}

输出格式：
Plan:
Step 1: #E1 = tool_name[input1]
Step 2: #E2 = tool_name[#E1, input2]
Step 3: #E3 = ...
"""

def planner_node(state: ReWOOState):
    response = llm.invoke(PLANNER_PROMPT.format(input=state["messages"][-1].content))
    return {"plan": response.content}

def worker_node(state: ReWOOState):
    """并行执行所有占位"""
    plan = state["plan"]
    evidence = {}
    
    # 解析计划，提取每个 #E
    import re
    steps = re.findall(r"(#E\d+)\s*=\s*(\w+)\[(.*?)\]", plan)
    
    for var, tool_name, args in steps:
        # 调用对应工具（简化：实际需要根据 tool_name 路由）
        tool = next((t for t in tools if t.name == tool_name), None)
        if tool:
            # 把 #E 占位替换成实际值
            resolved_args = args
            for e_var, e_val in evidence.items():
                resolved_args = resolved_args.replace(e_var, e_val)
            evidence[var] = tool.invoke(resolved_args)
        else:
            evidence[var] = f"工具 {tool_name} 不存在"
    
    return {"evidence": evidence}

def solver_node(state: ReWOOState):
    """合并计划 + 证据，生成答案"""
    prompt = f"""请基于以下计划和证据回答用户问题：

    计划：
    {state['plan']}
    
    证据：
    {state['evidence']}
    
    用户问题：{state['messages'][-1].content}
    """
    response = llm.invoke(prompt)
    return {"answer": response.content}

graph_builder = StateGraph(ReWOOState)
graph_builder.add_node("planner", planner_node)
graph_builder.add_node("worker", worker_node)
graph_builder.add_node("solver", solver_node)

graph_builder.add_edge(START, "planner")
graph_builder.add_edge("planner", "worker")
graph_builder.add_edge("worker", "solver")
graph_builder.add_edge("solver", END)

graph = graph_builder.compile()

优缺点#

五、模式 3：Reflexion（自我反思）#

核心思想#

让 Agent 完成后自我评估，错了就重做。

Reflexion 论文（Shinn et al., 2023）的核心发现：用自然语言反思错误，比单纯重试效果显著更好。

三层循环#

1. Actor（执行）：完成任务
2. Evaluator（评估）：判断结果是否正确
3. Self-Reflection（反思）：如果错了，写下"我哪里错了，下次怎么避免"
4. Loop：把反思加入记忆，重新尝试

流程图#

用户问题
   ↓
[Actor] 生成答案 v1
   ↓
[Evaluator] 评估答案 → 错/对
   ↓ 错
[Self-Reflection] 写下反思："我错了，因为..."
   ↓
[Actor] 基于反思生成答案 v2
   ↓
[Evaluator] 评估 → 对/错
   ↓ 对
最终答案

完整 LangGraph 实现#

class ReflexionState(TypedDict):
    messages: Annotated[list, add_messages]
    attempt: int
    current_answer: str
    reflection: str           # 上一次的反思
    evaluation: str           # 上一次的评估

ACTOR_PROMPT = """请回答以下问题：
{messages}

{freflection}"""

EVALUATOR_PROMPT = """请评估以下答案是否正确：

问题：{question}
答案：{answer}

评估标准：
1. 答案是否直接回答了问题？
2. 事实是否正确？
3. 推理是否合理？

输出格式：
VERDICT: correct | incorrect
REASON: <简短理由>"""

REFLECTOR_PROMPT = """你的上一次答案被评估为错误。

问题：{question}
错误答案：{answer}
评估理由：{evaluation}

请反思：
1. 我哪里错了？
2. 下次怎么避免？
3. 我应该采取什么不同的策略？

输出简洁的反思（200字内）。"""

def actor_node(state: ReflexionState):
    """生成或重新生成答案"""
    reflection = state.get("reflection", "")
    last_msg = state["messages"][-1].content
    
    prompt = ACTOR_PROMPT.format(
        messages=last_msg,
        freflection=f"\n\n上次反思：{reflection}" if reflection else ""
    )
    response = llm.invoke(prompt)
    return {
        "current_answer": response.content,
        "attempt": state.get("attempt", 0) + 1
    }

def evaluator_node(state: ReflexionState):
    """评估答案"""
    prompt = EVALUATOR_PROMPT.format(
        question=state["messages"][-1].content,
        answer=state["current_answer"]
    )
    response = llm.invoke(prompt)
    return {"evaluation": response.content}

def reflector_node(state: ReflexionState):
    """基于评估写反思"""
    prompt = REFLECTOR_PROMPT.format(
        question=state["messages"][-1].content,
        answer=state["current_answer"],
        evaluation=state["evaluation"]
    )
    response = llm.invoke(prompt)
    return {"reflection": response.content}

def should_continue(state: ReflexionState) -> str:
    if "VERDICT: correct" in state["evaluation"]:
        return "end"
    if state["attempt"] >= 3:
        return "end"
    return "reflect"

graph_builder = StateGraph(ReflexionState)
graph_builder.add_node("actor", actor_node)
graph_builder.add_node("evaluator", evaluator_node)
graph_builder.add_node("reflector", reflector_node)

graph_builder.add_edge(START, "actor")
graph_builder.add_edge("actor", "evaluator")
graph_builder.add_conditional_edges(
    "evaluator",
    should_continue,
    {"reflect": "reflector", "end": END}
)
graph_builder.add_edge("reflector", "actor")  # 反思后重做

graph = graph_builder.compile()

关键洞察：为什么自然语言反思有效？#

研究显示，让 LLM 用文字写出"我哪里错了"，比单纯让它"再试一次"有效得多。原因：

1. 强制显式化错误：LLM 必须在 prompt 里指出具体错误，不能模棱两可

2. 形成可复用的教训：反思可以存入长期记忆，下次遇到类似问题直接用

3. 避免重复错误：下一次 attempt 时，reflection 注入到 prompt，LLM 会避开同样的坑

论文中，Reflexion 在 HumanEval（代码生成基准）上从 ReAct 的 71% 提升到 91%——只用了三轮反思。

六、三种模式对比与选择#

决策树#

你的任务对准确率要求高吗？
├── 是 → 需要可重试吗？
│        ├── 是 → Reflexion（成本高但准确）
│        └── 否 → 单次 ReAct + Plan-and-Execute
│
└── 否 → 步骤之间有依赖吗？
         ├── 强依赖（必须串行） → Plan-and-Execute
         └── 弱依赖（可并行） → ReWOO（最省 token）

实战经验：

• 80% 的场景 Plan-and-Execute 就够了

• 需要批量处理大量数据时用 ReWOO

• 关键决策（如代码生成、医疗诊断）用 Reflexion

七、混合架构：实战中很少用纯模式#

生产环境的 Agent 往往是多种模式的混合：

Planner（Plan-and-Execute）
   ↓
Executor（ReAct + Tool Use）
   ↓
Evaluator（Reflexion）
   ↓
如果错 → 重新生成（Reflexion 重试）
   ↓
最终答案

典型例子：Devin（AI 软件工程师）的架构就是 Plan-and-Execute + Reflexion——先规划任务，再逐步执行，每步评估对错，错了反思重做。

八、和 L1-L6 的连接#

九、动手时间：亲手对比三种模式#

请你用一个真实复杂问题对比三种模式：

测试问题#

"分析 2025 年中国新能源汽车市场销量前三的品牌，给出它们的差异化策略"

步骤#

1. 用 Plan-and-Execute 实现：观察执行步骤的清晰度

2. 用 ReWOO 实现：对比 token 消耗（用 LangSmith 或简单计数）

3. 用 Reflexion 实现：故意在第一轮提供错误信息，看 Agent 能不能通过反思纠正

评分维度#

十、小结#

十一、Read next#

下一节 L8 元认知与 Agent 评估，我们学习：

• 元认知（Metacognition）：让 Agent 知道自己"知不知道"

• Agent 评估方法：怎么衡量 Agent 的好坏

• Benchmark 设计：怎么评估一个 Agent 系统

主要参考：

• Plan-and-Solve 论文 — Plan-and-Execute 的原始论文

• ReWOO 论文 — 推理与执行分离

• Reflexion 论文 — 自我反思机制

十二、Sources#

• src-001 — Microsoft AI Agents for Beginners · T1 · 07-planning-design

• src-002 — Hugging Face Agents Course · T1 · Course

• ReWOO 论文 · T1 · arXiv:2305.18323

• Reflexion 论文 · T1 · arXiv:2303.11381

💬 有疑问？ 试试问我："我的项目该选哪个 Planning 模式？"、"Reflexion 怎么避免无限重试？"、"ReWOO 在生产环境怎么落地？"——我会用代码或比喻为你解答。