Agent模式：ReAct、Plan-and-Execute、Multi-Agent协作

引言

想象一下，你是一位餐厅老板，生意火爆到后厨完全失控：传菜员（LLM）只知道把菜单丢给厨师，厨师（工具调用）做完一道菜就等着，没人统筹冷菜、热菜、甜品的出餐顺序。结果就是客人等到崩溃，厨房乱成一锅粥。

这就是当前很多LLM应用的缩影——单次调用、无状态、无规划。但真实的业务场景，比如“帮我对比这三份合同的风险条款并生成摘要报告”，远不是一次Prompt能搞定的。它需要拆解（读合同→提取条款→对比→生成报告）、决策（哪份合同先读？摘要用什么格式？）、多步工具调用（文件解析、向量检索、数据库查询、文本生成）。

业界已经沉淀出三种主流的Agent编排模式：ReAct（推理+行动交替）、Plan-and-Execute（先规划后执行）、Multi-Agent（多角色协作）。本文将从源码层面拆解这三者的本质差异，并给出可直接落地的代码示例。

核心概念：从“单线程”到“交响乐团”

生活类比

• ReAct 像一个经验丰富的急诊医生：观察病人症状（Observation）→ 判断病因（Thought）→ 开检查单（Action）→ 看检查结果（Observation）→ 再判断…每一步都依赖上一步的实时反馈，灵活但可能绕弯路。

• Plan-and-Execute 像工程项目经理：先召开启动会，画出完整的甘特图（Plan），然后交给施工队按图施工（Execute）。施工队遇到小问题可以微调，但大方向不变。效率高，但应对突发变化能力弱。

• Multi-Agent 像一家创业公司：CEO（规划Agent）拆解目标，CTO（技术Agent）负责实现，CFO（财务Agent）负责成本核算，HR（HR Agent）负责资源协调。每个Agent有独立人格、独立记忆、独立工具集，通过消息传递协作。

技术定义

| 模式 | 核心循环 | 状态管理 | 适用场景 |

|------|---------|---------|---------|

| ReAct | Thought → Action → Observation 循环 | 短期上下文窗口 | 需要动态推理、工具结果不可预知的场景 |

| Plan-and-Execute | 一次性规划 → 逐步执行 → 可选再规划 | 长期计划 + 短期执行栈 | 任务步骤明确、执行路径相对固定的场景 |

| Multi-Agent | 多角色并发/串行消息传递 | 每个Agent独立状态 + 共享消息总线 | 需要角色分离、权限隔离、并行处理的复杂任务 |

源码/原理深度分析：LangGraph 实现对比

LangGraph 是目前实现这三类模式最底层的框架，它把Agent定义为状态图（StateGraph），节点是函数，边是路由逻辑。

1. ReAct 的核心循环（LangGraph 源码视角）

LangGraph 的 ReAct 实现本质是一个 StateGraph，包含 agent 和 tools 两个节点，通过 should_continue 条件边形成循环：

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class AgentState(TypedDict):
    messages: Annotated[List, operator.add]  # 消息列表，自动累加

# 定义节点函数
def call_model(state: AgentState):
    # 调用 LLM，让它决定是回答问题还是调用工具
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def call_tool(state: AgentState):
    # 解析 LLM 返回的 tool_calls，执行工具，返回结果
    tool_calls = state["messages"][-1].tool_calls
    results = [execute_tool(tc) for tc in tool_calls]
    return {"messages": results}

# 条件路由：如果 LLM 没有生成 tool_calls，则结束；否则继续循环
def should_continue(state: AgentState):
    if not state["messages"][-1].tool_calls:
        return "end"
    return "continue"

# 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", call_tool)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"continue": "tools", "end": END})
graph.add_edge("tools", "agent")  # 工具结果回传给 agent

关键洞察：ReAct 的“思考-行动”循环在 LangGraph 中就是图的自环。状态通过 messages 字段不断累加，LLM 每次都能看到完整的历史。这种设计的代价是上下文窗口消耗巨大，一次复杂任务可能产生几万token的中间推理过程。

2. Plan-and-Execute 的架构差异

Plan-and-Execute 不是简单的循环，而是两阶段图：第一个节点生成完整计划（包含多个步骤），第二个节点循环执行计划中的步骤。

class PlanState(TypedDict):
    plan: List[str]          # 计划步骤列表
    current_step: int        # 当前执行到第几步
    results: List[str]       # 每步的结果
    past_steps: Annotated[List[tuple], operator.add]  # 历史步骤

def planner(state: PlanState):
    # 一次性生成完整计划
    plan_prompt = f"任务: {task}\n请生成步骤清单（每行一个步骤）"
    plan_text = llm.invoke(plan_prompt)
    steps = [s.strip() for s in plan_text.split("\n") if s.strip()]
    return {"plan": steps, "current_step": 0}

def executor(state: PlanState):
    # 执行当前步骤，更新状态
    step = state["plan"][state["current_step"]]
    result = execute_step(step, state["results"])  # 可访问之前步骤的结果
    return {
        "results": state["results"] + [result],
        "current_step": state["current_step"] + 1,
        "past_steps": [(step, result)]
    }

def should_continue_plan(state: PlanState):
    return "continue" if state["current_step"] < len(state["plan"]) else "end"

关键差异：ReAct 的“下一步做什么”是每步动态决策的，而 Plan-and-Execute 的“下一步”是预先定义的。前者灵活但token消耗大，后者高效但无法应对“计划赶不上变化”——如果第一步的结果推翻了第二步的前提，Plan-and-Execute 就会硬着头皮执行错误步骤。进阶方案是加入“重新规划”（Re-Plan）节点，但那是后话。

3. Multi-Agent 的消息传递机制

Multi-Agent 在 LangGraph 中是多个StateGraph的组合。每个子Agent有自己的状态和循环，Agent之间通过 Send API 进行异步消息传递。

from langgraph.constants import Send
from langgraph.graph import StateGraph

class MultiAgentState(TypedDict):
    task: str
    agent_outputs: dict  # 各Agent的输出汇总

# 子Agent 1：数据收集
def data_agent(state):
    data = query_database(state["task"])
    return {"agent_outputs": {"data": data}}

# 子Agent 2：分析
def analysis_agent(state):
    data = state["agent_outputs"]["data"]
    analysis = llm.invoke(f"分析以下数据: {data}")
    return {"agent_outputs": {"analysis": analysis}}

# 子Agent 3：报告生成
def report_agent(state):
    analysis = state["agent_outputs"]["analysis"]
    report = llm.invoke(f"基于分析生成报告: {analysis}")
    return {"agent_outputs": {"report": report}}

# 用 Send 实现并行分发
def spawn_agents(state):
    # 根据任务动态决定哪些Agent参与，并行启动
    return [
        Send("data_agent", state),
        Send("analysis_agent", state),
        Send("report_agent", state),
    ]

builder = StateGraph(MultiAgentState)
builder.add_node("data_agent", data_agent)
builder.add_node("analysis_agent", analysis_agent)
builder.add_node("report_agent", report_agent)
builder.add_conditional_edges("router", spawn_agents, ["data_agent", "analysis_agent", "report_agent"])

关键洞察：Multi-Agent 的核心价值在于隔离与并行。每个Agent可以有自己的Prompt模板、工具白名单、甚至独立的LLM模型（比如规划用GPT-4o，工具调用用Claude）。缺点是通信开销大，如果Agent间需要频繁交换中间结果，性能会急剧下降。

实战代码：三个可运行的完整示例

示例1：ReAct 模式——智能客服工单处理

"""
场景：用户提交工单，Agent需要判断问题类型，查询知识库，调用API处理，最后生成回复。
"""
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from typing import Dict

# 1. 定义工具
@tool
def search_knowledge_base(query: str) -> str:
    """搜索内部知识库，返回相关文档片段"""
    # 模拟知识库检索
    knowledge = {
        "密码重置": "用户可通过企业微信自服务门户重置密码，需验证手机号",
        "网络故障": "首先重启路由器，若仍无法上网请联系IT服务台分机1234",
        "软件安装": "标准软件可通过软件中心自助安装，特殊软件需提交审批单"
    }
    for key, value in knowledge.items():
        if key in query:
            return value
    return "未找到相关知识，建议转人工"

@tool
def create_ticket(priority: str, category: str) -> str:
    """在ITSM系统创建工单，返回工单号"""
    import uuid
    ticket_id = f"INC-{uuid.uuid4().hex[:8].upper()}"
    return f"已创建工单 {ticket_id}，优先级: {priority}，分类: {category}"

# 2. 初始化LLM和Agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
agent = create_react_agent(llm, [search_knowledge_base, create_ticket])

# 3. 运行Agent
def handle_ticket(user_input: str) -> Dict:
    result = agent.invoke({
        "messages": [{"role": "user", "content": user_input}]
    })
    return {
        "final_answer": result["messages"][-1].content,
        "steps": [m.content for m in result["messages"][1:-1] if m.type == "ai"]
    }

# 测试
if __name__ == "__main__":
    response = handle_ticket("我的电脑连不上公司WiFi，试了重启也不行，帮我处理一下")
    print("最终回复:", response["final_answer"])
    print("\n推理过程:")
    for step in response["steps"]:
        print(f"  → {step[:80]}...")

示例2：Plan-and-Execute 模式——市场调研报告生成

"""
场景：给定一个产品，Agent需要生成一份包含市场分析、竞品对比、定价建议的完整报告。
"""
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from typing import List, Dict
import json

class PlanExecuteAgent:
    def __init__(self, llm):
        self.llm = llm
        self.plan = []
        self.results = {}
        self.current_step = 0

    def create_plan(self, task: str) -> List[str]:
        """Step 1: 生成执行计划"""
        prompt = PromptTemplate.from_template(
            """为任务生成详细的执行计划，每行一个步骤，步骤要具体可执行。
任务: {task}
要求: 最多5个步骤，每个步骤必须包含明确的输入和输出。
输出格式: JSON数组，如 ["步骤1描述", "步骤2描述"]"""
        )
        response = self.llm.invoke(prompt.format(task=task))
        # 解析JSON计划
        plan = json.loads(response.content)
        self.plan = plan
        print(f"📋 计划生成: {plan}")
        return plan

    def execute_step(self, step: str, context: Dict) -> str:
        """Step 2: 执行单个步骤"""
        prompt = PromptTemplate.from_template(
            """执行以下步骤，并返回结果。
步骤: {step}
已有的上下文信息:
{context}

请直接输出步骤的执行结果，不要解释过程。"""
        )
        response = self.llm.invoke(prompt.format(
            step=step,
            context=json.dumps(context, ensure_ascii=False)[:2000]
        ))
        return response.content

    def re_plan_if_needed(self, step_result: str) -> bool:
        """Step 3 (可选): 判断是否需要重新规划"""
        # 简单策略：如果结果中包含"无法"、"错误"等关键词，触发重新规划
        if any(word in step_result for word in ["无法", "错误", "失败"]):
            print("⚠️ 检测到异常，重新规划中...")
            return True
        return False

    def run(self, task: str) -> str:
        # 阶段一：规划
        self.create_plan(task)

        # 阶段二：执行循环
        while self.current_step < len(self.plan):
            step = self.plan[self.current_step]
            print(f"\n🔧 执行步骤 {self.current_step+1}: {step}")

            # 收集之前的结果作为上下文
            context = {"task": task, "results": self.results}

            # 执行当前步骤
            result = self.execute_step(step, context)
            self.results[f"step_{self.current_step}"] = result
            print(f"✅ 步骤结果: {result[:100]}...")

            # 检查是否需要重新规划（简化版）
            if self.re_plan_if_needed(result):
                self.create_plan(task)  # 重新生成计划
                self.current_step = 0   # 从头执行
                continue

            self.current_step += 1

        # 阶段三：汇总最终报告
        final_prompt = PromptTemplate.from_template(
            """基于以下各步骤结果，生成最终报告。
任务: {task}
步骤结果:
{results}

请输出结构化的报告，包含标题、正文、结论。"""
        )
        final_response = self.llm.invoke(final_prompt.format(
            task=task,
            results=json.dumps(self.results, ensure_ascii=False, indent=2)
        ))
        return final_response.content

# 使用示例
if __name__ == "__main__":
    llm = ChatOpenAI(model="gpt-4", temperature=0.3)
    agent = PlanExecuteAgent(llm)
    report = agent.run("分析智能手表市场，给出新品牌进入建议")
    print("\n\n📄 最终报告:\n", report)

示例3：Multi-Agent 模式——跨部门协作的合同审查

"""
场景：一份合同需要法务（风险审查）、财务（成本评估）、技术（可行性评估）三个角色协作审查。
"""
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing import TypedDict, Dict, List
import asyncio

# 定义共享状态
class ContractReviewState(TypedDict):
    contract_text: str
    legal_findings: str = ""
    finance_findings: str = ""
    tech_findings: str = ""
    final_decision: str = ""

# 三个子Agent的Prompt模板
LEGAL_PROMPT = """你是资深法务专家。审查以下合同，找出风险条款。
合同内容: {contract}
输出格式:
- 风险等级: 高/中/低
- 风险条款列表: 
  1. 条款原文摘要 - 风险描述 - 修改建议"""

FINANCE_PROMPT = """你是财务总监。评估以下合同的经济合理性。
合同内容: {contract}
重点关注: 付款条件、违约金、成本分摊、税务条款。
输出格式:
- 成本评估: 合理/偏高/偏低
- 关键财务条款分析"""

TECH_PROMPT = """你是技术架构师。评估以下合同的技术可行性。
合同内容: {contract}
重点关注: 交付物技术要求、验收标准、技术支持范围。
输出格式:
- 技术可行性: 高/中/低
- 技术风险点"""

# 各Agent的处理函数
def legal_agent(state: ContractReviewState):
    llm = ChatOpenAI(model="gpt-4", temperature=0.2)
    response = llm.invoke(LEGAL_PROMPT.format(contract=state["contract_text"]))
    return {"legal_findings": response.content}

def finance_agent(state: ContractReviewState):
    llm = ChatOpenAI(model="gpt-4", temperature=0.2)
    response = llm.invoke(FINANCE_PROMPT.format(contract=state["contract_text"]))
    return {"finance_findings": response.content}

def tech_agent(state: ContractReviewState):
    llm = ChatOpenAI(model="gpt-4", temperature=0.2)
    response = llm.invoke(TECH_PROMPT.format(contract=state["contract_text"]))
    return {"tech_findings": response.content}

# 汇总决策节点
def final_review(state: ContractReviewState):
    llm = ChatOpenAI(model="gpt-4", temperature=0.2)
    summary_prompt = """你是CEO，综合以下三个部门的意见，做出最终决策。
法务意见: {legal}
财务意见: {finance}
技术意见: {tech}

请输出:
1. 综合风险评估
2. 是否批准签署（批准/有条件批准/拒绝）
3. 附加条件（如有）"""
    response = llm.invoke(summary_prompt.format(
        legal=state["legal_findings"],
        finance=state["finance_findings"],
        tech=state["tech_findings"]
    ))
    return {"final_decision": response.content}

# 构建多Agent图
def build_multi_agent_graph():
    graph = StateGraph(ContractReviewState)

    # 添加节点
    graph.add_node("legal", legal_agent)
    graph.add_node("finance", finance_agent)
    graph.add_node("tech", tech_agent)
    graph.add_node("ceo_review", final_review)

    # 并行执行三个子Agent（无依赖关系）
    graph.set_entry_point("legal")
    graph.add_edge("legal", "finance")
    graph.add_edge("finance", "tech")
    graph.add_edge("tech", "ceo_review")
    graph.add_edge("ceo_review", END)

    return graph.compile()

# 运行
if __name__ == "__main__":
    contract_sample = """本合同约定甲方支付预付款50%，验收合格后支付剩余50%。违约金为合同总额的30%。交付物需包含完整的API接口文档及源代码。技术验收标准为系统可用性99.9%。"""

    app = build_multi_agent_graph()
    final_state = app.invoke({"contract_text": contract_sample})

    print("="*50)
    print("🤖 法务审查结果:\n", final_state["legal_findings"])
    print("="*50)
    print("💰 财务审查结果:\n", final_state["finance_findings"])
    print("="*50)
    print("🛠️ 技术审查结果:\n", final_state["tech_findings"])
    print("="*50)
    print("🏛️ CEO最终决策:\n", final_state["final_decision"])

方案对比：业界实现全景

graph TD A[Agent编排模式] --> B[ReAct] A --> C[Plan-and-Execute] A --> D[Multi-Agent] B --> B1[LangChain AgentExecutor] B --> B2[LangGraph create_react_agent] B --> B3[AutoGPT] C --> C1[LangChain PlanAndExecute] C --> C2[BabyAGI] C --> C3[LangGraph 手动实现] D --> D1[AutoGen] D --> D2[CrewAI] D --> D3[LangGraph Multi-Agent] B1 -.->|"简单但笨重"| B2 C1 -.->|"计划易过期"| C3 D1 -.->|"对话驱动"| D2

| 框架 | 模式支持 | 状态持久化 | 可视化 | 学习曲线 | 生产就绪度 |

|------|---------|-----------|--------|---------|-----------|

| LangGraph | 全部支持 | ✅ Checkpointer | ✅ 原生 | 陡峭 | 高 |

| AutoGen | Multi-Agent为主 | ✅ 对话历史 | ❌ | 中等 | 中 |

| CrewAI | Multi-Agent为主 | ❌ | ❌ | 平缓 | 中 |

| LangChain AgentExecutor | ReAct为主 | ❌ | ❌ | 平缓 | 低（易踩坑） |

| BabyAGI | Plan-Execute | ❌ | ❌ | 中等 | 低 |

关键结论：LangGraph 是唯一一个能在同一个框架内优雅实现三种模式的工具，且支持Checkpoint持久化、人工介入（Human-in-the-loop）、流式输出等生产级特性。如果你已经决定用LangChain生态，直接上LangGraph是正解。

最佳实践与避坑指南

最佳实践

1. 混合模式：不要死守一种模式。推荐“外Plan内ReAct”——外层用Plan-and-Execute规划大步骤，每个步骤内部用ReAct动态决策。LangGraph的嵌套图可以完美支持。

1. 状态最小化：Multi-Agent模式下，Agent间传递的消息要精简。传“结论”而不是传“全部推理过程”，否则token消耗会指数级增长。

1. 工具调用容错：真实环境中的工具（API、数据库）可能失败。一定要在工具层加 try-catch，返回结构化错误信息，让LLM能理解错误并尝试替代方案。

1. 人工审批闸门：对于关键操作（如支付、删除数据），在LangGraph中用 interrupt_before 注入人工审批节点。这是生产级Agent的必备能力。

常见坑

| 坑 | 现象 | 解决方案 |

|----|------|---------|

| 上下文爆炸 | 请求耗时从2s涨到30s，成本翻10倍 | 使用 trim_messages 压缩中间步骤，或切换到Plan模式 |

| 工具循环死锁 | Agent反复调用同一个失败的工具 | 设置最大迭代次数，并在工具错误信息中明确提示“不要再调用此工具” |

| Multi-Agent死锁 | Agent A 等 B 的结果，B 等 A 的结果 | 使用 Send API实现异步消息，避免同步等待；设置超时机制 |

| 计划过期 | Plan执行到一半，外部条件已变化 | 在Plan模式中加入“重新规划”节点，每N步检查一次计划有效性 |

| 模型幻觉 | Agent编造工具返回结果 | 严格要求工具返回结构化为JSON，并在Prompt中强调“只可引用真实返回的数据” |

总结与延伸思考

三种模式本质上是“思考密度”和“规划深度”的权衡：

• ReAct 是“走一步看一步”的游击队，适合探索未知问题，但效率低、成本高。

• Plan-and-Execute 是“先画图纸再施工”的工程队，适合流程稳定的重复任务，但应变能力弱。

• Multi-Agent 是“多兵种协同”的集团军，适合复杂分工任务，但指挥通信成本高。

延伸思考：未来的Agent编排会走向自适应模式——让LLM自己根据任务复杂度动态选择ReAct、Plan或Multi-Agent。LangGraph已经提供了 create_agent 的配置化入口，这将是Agent框架的下一波演进方向。

最后，无论你选择哪种模式，记住这条铁律：Agent的价值不在“会思考”，而在“会正确地使用工具”。 花在工具设计和异常处理上的时间，永远比调Prompt更值得。