第2章：Agent 控制流模型 —— ReAct / Plan-Execute / Reflexion / Graph

控制流是 agent 的”思考节奏”——LLM 应该看一步走一步(ReAct),还是先把大计划做出来再分头执行(Plan-Execute)?要不要给它”反悔重来”的能力(Reflexion)?复杂业务该不该把每一步都画成图(LangGraph DAG)?本章用 5 种主流控制流模型把这些问题答清楚,每种都给出代码骨架、典型 prompt 和适用场景判断,让你看到一个 agent 任务能立刻知道”该用哪种 control flow”。

📑 目录

1. 为什么要分类控制流
2. ReAct:Reason + Act 循环
3. Plan-and-Execute:先 plan 后 execute
4. Reflexion:自反思改写计划
5. DAG / State Machine:LangGraph 范式
6. Tool-as-Thought:用 tool call 表达推理
7. 选型决策树
自我检验清单
参考资料

1. 为什么要分类控制流

控制流决定:

问题	不同控制流的答案
谁决定下一步做什么?	LLM 即时决定(ReAct) / 提前 plan(Plan-Exec) / 状态机硬编码(Graph)
中间能不能反悔?	不行(ReAct/Graph) / 可以(Reflexion)
能不能并行?	难(ReAct) / 可以(Plan-Exec, Graph)
失败怎么处理?	重试当前 step / 重新 plan / state transition
LLM 调用次数?	多(每步一次) / 少(plan 阶段集中)

🌟 错误的控制流 = 错误的工程基础。先选对范式再写代码,比”先写 ReAct,后面发现不够再改”省 10 倍时间。

2. ReAct:Reason + Act 循环

Yao et al., “ReAct: Synergizing Reasoning and Acting in Language Models”, arXiv 2210.03629

2.1 思想

把”推理(Thought)“和”行动(Action)“显式交错——每一步 LLM 先想,再决定调哪个 tool,看 tool 结果(Observation),再思考下一步:

Thought: 用户想知道北京天气,我需要调天气 API
Action: weather_api(city="北京")
Observation: 24°C, 多云
Thought: 用户可能还关心未来几天,但只问了今天,直接回答
Action: respond("北京今天 24°C 多云,适合出行")

2.2 代码骨架(LangGraph)

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def weather_api(city: str) -> str:
    """查询城市当前天气"""
    return "24°C, 多云"

@tool
def search(query: str) -> str:
    """通用搜索"""
    ...

agent = create_react_agent(
    ChatOpenAI(model="gpt-4o"),
    tools=[weather_api, search],
)

result = agent.invoke({"messages": [("user", "北京天气怎么样?")]})

LangGraph 内置 create_react_agent 是工业级 ReAct 实现,推荐起手。

2.3 典型 prompt

You are a helpful assistant. Use the following format:

Thought: think about what to do
Action: tool_name(args)
Observation: tool result
... (repeat)
Thought: I have enough info
Action: respond(answer)

OpenAI / Anthropic 模型已经内置 tool-calling 范式,实际写时不用手拼这个 format,直接给 tool schema 即可。

2.4 适用场景

✅ 简单的多步任务(查信息、算东西、做决策) ✅ 不需要长期规划 ✅ 每步可以独立验证

❌ 不适用于:需要并行、需要严格事务、超过 10 步的复杂任务

2.5 失败模式

死循环:LLM 反复调同一个工具——加 max_iterations 限制(典型 10-15)
遗忘上下文:历史 turn 太多,LLM 漏看关键 observation——用 Memory 摘要(模块五)
Tool 选错:tool 描述模糊或太多——精简到 < 10 个,描述要清晰

3. Plan-and-Execute:先 plan 后 execute

3.1 思想

ReAct 是”边走边想”,Plan-Execute 是”先画地图再上路”:

Phase 1: Planner LLM 根据任务生成完整 plan
  → "1. 查 A; 2. 用 A 的结果查 B; 3. 把 A、B 给 LLM 总结"

Phase 2: Executor 按 plan 顺序执行
  → 调 A → 调 B → 总结

Phase 3:(可选)Reviewer 检查结果,不行就回到 Phase 1 重 plan

3.2 代码骨架

from langgraph.graph import StateGraph
from typing import TypedDict, List

class State(TypedDict):
    task: str
    plan: List[str]
    completed: List[dict]
    final_answer: str

def planner(state):
    plan = llm.invoke(f"Make a plan for: {state['task']}")
    return {"plan": parse_steps(plan)}

def executor(state):
    if not state["plan"]:
        return {"final_answer": llm.summarize(state["completed"])}
    next_step = state["plan"][0]
    result = execute_step(next_step)
    return {
        "plan": state["plan"][1:],
        "completed": state["completed"] + [{"step": next_step, "result": result}],
    }

graph = StateGraph(State)
graph.add_node("planner", planner)
graph.add_node("executor", executor)
graph.set_entry_point("planner")
graph.add_edge("planner", "executor")
graph.add_conditional_edges(
    "executor",
    lambda s: "executor" if s["plan"] else END,
)

3.3 优势

可并行:plan 中独立的 step 可以并发执行
省 LLM:plan 阶段一次性想清楚,不用每步都 LLM
可审计:plan 是显式的,可以让人审批后再执行

3.4 适用场景

✅ 任务可以提前规划(数据 ETL、报告生成、批量处理) ✅ 需要并行加速 ✅ 需要 human-in-the-loop 审批 plan

❌ 不适用于:动态环境(plan 频繁失效)、单步任务

4. Reflexion:自反思改写计划

Shinn et al., “Reflexion: Language Agents with Verbal Reinforcement Learning”, arXiv 2303.11366

4.1 思想

ReAct 失败一次就死了。Reflexion 的核心:失败后让 LLM 自己写”反思”,作为下一次尝试的额外输入:

Trial 1: 尝试解决任务 → 失败/低分
  ↓
Reflexion: LLM 反思"我哪里做错了?为什么?下次该怎么做?"
  → 写入 reflective memory
  ↓
Trial 2: 把 reflection 加到 prompt,再尝试 → 通常表现更好

不需要梯度更新——这就是”verbal reinforcement learning”的精髓。

4.2 代码骨架

def run_with_reflexion(task, max_trials=3):
    reflections = []
    for trial in range(max_trials):
        prompt = task
        if reflections:
            prompt += f"\n\nPast reflections:\n" + "\n".join(reflections)
        
        result = react_agent(prompt)
        score = evaluate(result, task)
        
        if score >= 0.8:
            return result
        
        # 失败,反思
        reflection = llm.invoke(
            f"Task: {task}\nMy attempt: {result}\nIt scored {score}. "
            f"Why did I fail? What should I do differently next time?"
        )
        reflections.append(reflection)
    
    return result  # 最后一次的结果

4.3 适用场景

✅ 有评估信号(代码能否编过、单测是否通过、用户反馈) ✅ 任务允许多次尝试 ✅ 需要 agent 自我改进(典型:HumanEval 代码生成、ALFWorld 任务)

❌ 不适用:无法评估对错的开放性任务、不允许重试的实时场景

4.4 与 LangMem procedural memory 的关系

Reflection 写入的内容可以视为 Procedural Memory(模块五),会改善未来同类任务的表现——这就是 LangMem 的 create_prompt_optimizer 在做的事。

5. DAG / State Machine:LangGraph 范式

5.1 思想

把 agent 行为画成显式的有向图,节点是”做什么”,边是”什么条件下走哪条”:

       ┌──────────┐
       │  start   │
       └────┬─────┘
            │
       ┌────▼─────┐
       │ classify │  根据用户意图分类
       └─┬──┬──┬──┘
         │  │  │
   ┌─────┘  │  └─────┐
   ▼        ▼        ▼
┌─────┐ ┌──────┐ ┌──────┐
│ FAQ │ │ Plan │ │ Tool │
└──┬──┘ └──┬───┘ └──┬───┘
   │       │        │
   └───────┴────────┘
            │
       ┌────▼─────┐
       │ respond  │
       └──────────┘

5.2 LangGraph 代码

from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal

class State(TypedDict):
    messages: list
    intent: str
    answer: str

def classify(state):
    intent = llm.classify(state["messages"][-1].content)
    return {"intent": intent}

def faq(state):
    return {"answer": faq_lookup(state["messages"][-1].content)}

def plan_route(state):
    plan = llm.plan(state["messages"][-1].content)
    return {"answer": execute_plan(plan)}

def tool_route(state):
    return {"answer": call_tool(state["messages"][-1].content)}

def respond(state):
    return {"messages": state["messages"] + [{"role": "assistant", "content": state["answer"]}]}

graph = StateGraph(State)
for name, fn in [("classify", classify), ("faq", faq), ("plan", plan_route), ("tool", tool_route), ("respond", respond)]:
    graph.add_node(name, fn)

graph.set_entry_point("classify")
graph.add_conditional_edges(
    "classify",
    lambda s: s["intent"],   # 路由依据
    {"faq": "faq", "plan": "plan", "tool": "tool"},
)
for n in ["faq", "plan", "tool"]:
    graph.add_edge(n, "respond")
graph.add_edge("respond", END)

app = graph.compile()

5.3 优势

可视化 + 可审计:每条路径都画在图上,review 容易
支持 cycle(LangGraph)和 conditional edge
天然适合 checkpointer + Saga(下章详讲)
多 agent supervisor / swarm 都是 graph 的特例

5.4 缺点

冷启动成本:简单任务也要画图,过度工程
LLM 不参与”形状”决策:图是固定的,LLM 只在节点内自由

5.5 何时用 Graph 而非 ReAct

信号	选 Graph
业务流程相对固定(>80% 路径在已知集合)	✅
需要 human-in-the-loop 中断	✅
需要 durable execution	✅
多 agent 协作	✅
任务高度动态、每次路径都不同	❌(选 ReAct)

🌟 生产 agent 中,Graph 是工业首选——LangGraph 在 2026 GitHub stars 反超 CrewAI 就是这个原因。

6. Tool-as-Thought:用 tool call 表达推理

6.1 思想

新趋势:LLM 内置思考能力(o1、Claude Extended Thinking、DeepSeek-R1)后,推理本身就是模型内部的事,外部 agent 只用关心”调什么 tool”。

User: 帮我订北京到上海的高铁,明天早上 9 点左右

LLM 内部思考(隐式)...

Tool calls:
  1. search_train(from="北京", to="上海", date="2026-05-06", time_window="08:00-10:00")
  2. user_confirm(options=...)
  3. book(train_id=..., user_id=...)

agent 的”显式控制流”只是把 LLM 决定的 tool call 顺序串起来——thinking 已经被模型 builtin。

6.2 OpenAI Agents SDK / Swarm 的极简 handoff

from agents import Agent, handoff

booking = Agent(
    name="Booking Agent",
    instructions="处理订票",
    tools=[search_train, book],
)

triage = Agent(
    name="Triage",
    instructions="判断用户意图,handoff 给对应 agent",
    handoffs=[booking, ...],
)

result = triage.run("订北京到上海的票")

OpenAI 的设计哲学:让 LLM 自己决定 handoff,不要硬编码 graph。极简代码,适合 LLM 推理能力够强的场景。

6.3 适用场景

✅ 使用 o1 / R1 / Claude Extended Thinking 等强推理模型 ✅ 业务路径高度动态 ✅ 不需要严格审计每一步路由

❌ 不适用:模型 reasoning 弱、需要 deterministic 控制、合规审计严格

7. 选型决策树

你的 agent 任务复杂度?
│
├─ 简单(< 5 步,无并行)
│   └─ ReAct(LangGraph create_react_agent)
│
├─ 中等(可提前规划,需要并行)
│   └─ Plan-and-Execute
│
├─ 复杂(业务流程相对固定,多 agent,需要 durable)
│   └─ DAG / State Machine(LangGraph)
│
├─ 超复杂(高度动态,LLM 推理强)
│   └─ Tool-as-Thought(OpenAI Agents SDK / Swarm)
│
└─ 任务可重试且有评估信号
    └─ + Reflexion 包一层(独立维度,可叠加上面任一种)

🍎 常见组合:

客服 agent = Graph(主流程)+ ReAct(单 turn 内的工具调用)
代码 agent(Cursor / Devin) = Plan-Execute + Reflexion(测试失败重试)
多 agent 系统 = Graph(supervisor)+ ReAct(每个 worker 内部)
创意写作 agent = Tool-as-Thought(用 R1)+ Reflexion(用户评分驱动)

✅ 自我检验清单

5 种控制流:能默写 ReAct / Plan-Execute / Reflexion / Graph / Tool-as-Thought 的核心思想
ReAct 死循环:能解释为什么要 max_iterations,以及调多大合适
Plan-Execute 优势:能给 3 个具体场景,Plan-Execute 比 ReAct 显著好
Reflexion 必要条件:能列出 Reflexion 适用的 3 个前提
LangGraph 代码:能不看资料写一个简单 classify-route 的 graph
Graph 何时用:能用 5 个信号判断是否该上 Graph
Tool-as-Thought:能解释为什么 o1 时代很多 agent 反而变简单了
常见组合:能为 3 个具体业务设计”组合控制流”方案
失败模式:每种控制流能列出 2 种典型失败和应对

📚 参考资料

论文

ReAct (Yao et al., 2022):arXiv 2210.03629
Reflexion (Shinn et al., 2023):arXiv 2303.11366
Plan-and-Solve Prompting (Wang et al., 2023):arXiv 2305.04091
Tree of Thoughts (Yao et al., 2023):arXiv 2305.10601 —— Graph 思想的延伸
Voyager (Wang et al., 2023):Minecraft agent,procedural memory + 自动 skill 演化

框架文档

LangGraph create_react_agent:docs.langchain.com
LangGraph StateGraph:官方教程
OpenAI Agents SDK:openai.github.io/openai-agents-python
Pydantic AI Agents:ai.pydantic.dev
AWS Prescriptive Guidance: Reasoning Patterns:AWS Docs

中文解读

猛猿:ReAct/Reflexion/Plan-Execute 详解 —— 知乎专栏
方佳瑞:LangGraph 控制流实战