从RAG到Agent：LLM应用架构的范式跃迁

📍 本文是「LLM进阶：从会用到底层精通」专题的第 5/10 篇

📊 难度：进阶 | ⏱️ 预计阅读：20 分钟

学习目标

🎯 学完本文后，你将能够：

- 清晰区分 RAG 和 Agent 的能力边界：一个负责"知"，一个负责"行"

- 掌握 Agent 的四种核心设计模式：ReAct、Plan-Execute、Multi-Agent、Human-in-the-Loop

- 理解 MCP 与 A2A 协议的设计目标、适用场景及互补关系

- 用 LangGraph 实现一个完整的 ReAct Agent，完成天气搜索+空气质量查询任务

1. 为什么RAG不够用了？

2024年，RAG 几乎成了 LLM 落地的标配——给客服搭知识库问答、给法务做合同审查、给企业做内部文档搜索。RAG 本质上给 LLM 配了一本"参考书"：用户问什么，系统先去查，把相关内容贴到 prompt 里，模型照着答题。

但这套方案有一个致命局限：RAG 只能告诉用户"答案是什么"，不能帮用户"把事情办了"。

设想一个真实场景：用户对着旅行助手说——「帮我订一张明天从上海飞北京的机票，要下午的航班，顺便查一下北京明天的空气质量，如果 PM2.5 超标就提醒我带口罩。」

用纯 RAG 能做什么？检索航班的政策文档和订票流程说明，告诉用户「您可以通过以下步骤订票……」。然后呢？票没订，天气没查——用户得到了一堆"说明书"，但事还得自己干。这就像你问朋友"怎么订机票"，朋友给了你一个操作手册，而不是帮你把票订了。

Agent 则完全不同。同一个需求交给 Agent：先调用机票 API 搜索符合条件的航班，再调用天气 API 查北京空气质量，然后综合判断 PM2.5 是否超标，最后给出整合建议——「已为您找到 3 个符合条件航班，北京明天 PM2.5 预计 85μg/m³，建议佩戴口罩。」全程一个闭环，用户只差一个确认按钮。

✨ 一句话记住：RAG 给模型的是上下文（Context），Agent 给模型的是行动能力（Agency）。前者让模型会查资料，后者让模型会干活。

这个转变有多大？举一个量化的对比。传统 RAG 的典型架构是：用户输入 → Embedding → 向量检索 → Prompt 拼接 → LLM 生成。每一步都是线性的、可预测的。Agent 的架构则变成了：用户输入 → 模型自主规划 → 选择工具 → 执行 → 观察结果 → 再规划 → ... 直到任务完成。控制权从"你写死规则"变成了"模型自己决定"——这是从程序化到自主化的跃迁。

但 Agent 不是银弹。它的代价也很直白：延迟更高、成本更大、不可控性更强。一个 RAG 问答可能 500ms 返回，一个多步 Agent 可能要跑 20-30 秒，token 消耗是 RAG 的 5-10 倍。更关键的是，Agent 出错时不是答错了——是真把事情办错了，比如用错误的价格下了单、给错误的人发了邮件。根据 LangChain 2025 年度报告，Agent 在企业中的投产率从 2024 年的不到 10% 飙升到了 57%，但 Gartner 也预测 40% 的 Agentic AI 项目会在 2027 年前失败——核心原因就是对这些代价的低估。

2. Agent的设计模式全景

Agent 不是一种单一技术，而是一族设计模式的集合。下面按实际工程使用频率从高到低逐一拆解。

2.1 ReAct：Agent的心跳

核心思想：让模型"想一步，做一步，看一眼结果，再想下一步"。这就是 2022 年 Google Research 提出的 ReAct（Reasoning + Acting）范式——论文标题直译就是"让语言模型中的推理与行动协同工作"。

ReAct 的循环非常简单：

你可以把它理解成棋手下棋——你不可能下一步就交卷，你得看对方怎么应对，再想下一步。ReAct 把这个"边想边做"的过程变成了 Agent 的基础节拍器。

🛠️ 实战经验：ReAct 在生产环境里最容易出的问题是无限循环——模型偶尔会卡在"调工具→不满意→再调→还是不满意→再调"的死循环里。解法不是调 prompt，而是在代码里设一个 max_iterations 硬上限，比如 15 步，超过就直接走 fallback。另外，小模型（7B-13B 级别）跑 ReAct 稳定性很差——很多连"什么时候该停"都判断不准。生产环境建议用 70B+ 模型或 GPT-4o/Claude 级别。还有一个容易被忽视的细节：ReAct 的 Thought 部分是"写给模型自己看的"——它不需要也不应该展示给用户，这属于内部推理过程，类似于 o1 的 hidden CoT。

2.2 Plan-Execute：先画地图再上路

核心思想：ReAct 是边走边想，Plan-Execute 是先规划再执行。模型先别急着动手，把整个任务拆成 1-2-3-4 步的完整计划，然后一步步照计划执行。

为什么需要 Plan-Execute？因为 ReAct 在长链任务中有个致命问题——模型会"忘记初衷"。一个需要 8 步才能完成的任务，ReAct 跑到第 5 步时模型可能已经偏离了最初的目标。Plan-Execute 提前把总体规划固定下来，相当于给模型装了一个"行动纲领"。

代价也很明显：如果初始计划本身就有问题，后面全白干。而且 Plan-Execute 的首次响应延迟比 ReAct 高——必须先等计划生成完才能动手。

🛠️ 实战经验：Plan-Execute 在"任务步骤可预测"的场景里极好，比如代码审查流水线。但在"需要实时反馈调整"的场景反而不如 ReAct。实际生产里我们常用混合方案——Plan-Execute 做主框架，每个步骤内跑 ReAct。

2.3 Multi-Agent：一个人搞不定就组团队

核心思想：把大任务拆给多个"专业 Agent"，每个只干自己擅长的事——比如一个 Agent 写代码、另一个做安全检查、第三个写测试。

三种典型协作模式：

🛠️ 实战经验：一个残酷的真相——大多数场景下，单个强模型跑 ReAct 就够了。我见过太多团队一上来就搞 3-5 个 Agent 协作，token 成本飙升，debug 难度爆炸，最后效果还不如一个用心写 prompt 的单 Agent。什么时候真正需要 Multi-Agent？只有当任务天然需要不同的"专业视角"时——这些是真正的分工，不是人为拆出来的。

✨ 一句话记住：不到万不得已别上 Multi-Agent——一个强模型+好工具比五个弱 Agent 强十倍。

2.4 Human-in-the-Loop：别让Agent替你担责

核心思想：Agent 在执行"写操作"（发送消息、修改数据、执行交易）之前暂停，把决定权交还给人。

这不是 Agent 能力不够的妥协——它是生产级系统的必要组件。关键设计点在于控制粒度：每个工具调用都让人确认一遍，那还不如人自己干。我们线上只对"写操作"设 HITL，"读操作"全自动。

🛠️ 实战经验：HITL 的等待时间必须有超时策略——如果用户 30 分钟没确认，是自动取消还是降级处理？这个逻辑在设计阶段就要定好。

✨ 一句话记住：Agent 可以帮你做决策提案，但不可逆的"写操作"必须让人签字——这不是技术的限制，是责任的边界。

3. MCP与A2A：Agent的连接标准

讲完 Agent 怎么"想"，来聊聊 Agent 怎么"连"。2025 年，Anthropic 的 MCP 和 Google 的 A2A 两个协议基本定义了未来 Agent 生态的基础设施。

3.1 MCP：让LLM调用任何工具

MCP（Model Context Protocol）就是给 AI 模型装上"USB-C 接口"。以前每个工具都要单独写适配代码——10 个工具配 10 个模型就是 100 套代码。MCP 把这个变成"一头插模型，一头插工具"——10+10=20 套就够了。

MCP 采用 JSON-RPC 2.0 通信，定义了 Host（AI 应用）和 Server（工具提供方）之间的标准化交互。它由三个核心原语构成：Tools（模型可调用的函数，模型控制）、Resources（模型可读取的数据，应用控制）、Prompts（预定义的 prompt 模板，用户控制）。这三个原语覆盖了 Agent 与外部世界交互的完整需求。

MCP 的设计灵感直接来自 LSP（Language Server Protocol）——LSP 让任意 IDE 连任意语言服务器，MCP 让任意 AI 应用连任意工具服务器。这个类比非常精准：就像 VS Code 不需要为 Python、Go、Rust 各写一套语言支持，Agent 也不需要为每个数据库、API、文件系统各写一套集成代码。

截止 2025 年底，已有近 2000 个 MCP Server 注册在案，覆盖 GitHub、Notion、Stripe、Postman、Hugging Face 等主流服务。2025 年 12 月 MCP 被捐赠给 Linux 基金会下属的 Agentic AI Foundation，由 Anthropic、Block 和 OpenAI 共同管理——它已经从一个"Anthropic 的协议"变成了行业标准。

💡 关键要点：MCP 不限模型、不限供应商——你可以用 MCP Server 给 GPT-4o、Claude、Gemini 甚至本地部署的开源模型提供服务。

3.2 A2A：让Agent之间互相通信

如果 MCP 解决了"一个人怎么用工具"，A2A（Agent-to-Agent Protocol）则解决"多个人怎么配合干活"。

A2A 基于 HTTP + JSON-RPC + SSE（Server-Sent Events），定义了 Agent 之间的能力发现、任务委派、消息传递和结果交换的标准化方式。核心概念：

A2A 的一个关键设计是支持流式通信（SSE）——长任务不需要等全部完成才返回结果，可以实时推送中间状态和部分结果。这对多 Agent 协作场景至关重要：主 Agent 不需要 block 等子 Agent 跑完，可以并行管理多个子任务。

🛠️ 实战经验：A2A 目前仍处于早期阶段（2026 年 5 月刚发布 v1.0），生态远不如 MCP 成熟。小团队现阶段可以先不投 A2A——在单应用内做多 Agent 协作，自己的编排逻辑就够了。A2A 的真正价值在跨组织、跨系统的 Agent 互操作，这个场景在 2026 下半年才会逐步出现。

MCP 和 A2A 的分工一句话记：MCP 是纵向的（Agent ↔ 工具），A2A 是横向的（Agent ↔ Agent）。两者互补，不是竞争。

4. 代码实践：用LangGraph实现ReAct Agent

下面用 LangGraph 实现一个能"搜索天气 → 查空气质量 → 综合分析 → 回答用户"的 ReAct Agent。

任务描述：用户问「北京今天天气怎么样？空气质量如何？适合户外运动吗？」——Agent 需要先查天气、再查空气质量、最后综合给建议。

import os
from typing import Annotated, Sequence, TypedDict, Literal
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import BaseMessage, ToolMessage, HumanMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver

# Step 1: 定义Agent状态——只需维护消息列表
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]

# Step 2: 定义工具
@tool
def search_weather(city: str) -> str:
    """查询指定城市今天的天气。参数 city 为城市名称。"""
    weather_db = {
        "北京": "北京今天晴，温度 15°C ~ 28°C，湿度 35%，风力 2-3级",
        "上海": "上海今天多云转阴，温度 22°C ~ 31°C，湿度 70%，风力 3-4级",
    }
    return weather_db.get(city, f"未找到{city}的天气数据")

@tool
def search_air_quality(city: str) -> str:
    """查询指定城市的空气质量指数(AQI)和PM2.5。参数 city 为城市名称。"""
    aqi_db = {
        "北京": "北京 AQI 68，PM2.5 35μg/m³，PM10 78μg/m³，空气质量：良",
        "上海": "上海 AQI 45，PM2.5 22μg/m³，PM10 52μg/m³，空气质量：优",
    }
    return aqi_db.get(city, f"未找到{city}的空气质量数据")

tools = [search_weather, search_air_quality]
tools_by_name = {t.name: t for t in tools}

# Step 3: 绑定工具到模型
model = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)

# Step 4: 定义图节点
def call_model(state: AgentState) -> AgentState:
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def call_tool(state: AgentState) -> AgentState:
    last_message = state["messages"][-1]
    tool_messages = []
    for tool_call in last_message.tool_calls:
        tool_func = tools_by_name[tool_call["name"]]
        result = tool_func.invoke(tool_call["args"])
        tool_messages.append(
            ToolMessage(content=str(result), tool_call_id=tool_call["id"])
        )
    return {"messages": tool_messages}

def should_continue(state: AgentState) -> Literal["call_tool", END]:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "call_tool"
    return END

# Step 5: 构建并编译图
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("call_tool", call_tool)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {
    "call_tool": "call_tool",
    END: END,
})
workflow.add_edge("call_tool", "agent")
agent = workflow.compile(checkpointer=MemorySaver())

# 运行Agent
config = {"configurable": {"thread_id": "demo-1"}}
task = HumanMessage(content="北京今天天气怎么样？空气质量如何？适合户外运动吗？")

print("=== ReAct Agent 执行轨迹 ===\n")
for event in agent.stream({"messages": [task]}, config=config):
    for node_name, node_output in event.items():
        if node_name == "agent":
            msg = node_output["messages"][-1]
            if hasattr(msg, "tool_calls") and msg.tool_calls:
                for tc in msg.tool_calls:
                    print(f"  🔧 调用工具: {tc['name']}({tc['args']})")
            elif hasattr(msg, "content") and msg.content:
                print(f"  💬 模型回复: {msg.content[:150]}...")
        elif node_name == "call_tool":
            for m in node_output["messages"]:
                print(f"  📊 工具返回: {m.content[:120]}...")

import os

from typing import Annotated, Sequence, TypedDict, Literal

from langchain_openai import ChatOpenAI

from langchain_core.tools import tool

from langchain_core.messages import BaseMessage, ToolMessage, HumanMessage

from langgraph.graph import StateGraph, END

from langgraph.graph.message import add_messages

from langgraph.checkpoint.memory import MemorySaver

Step 1: 定义Agent状态——只需维护消息列表

class AgentState(TypedDict):

messages: Annotated[Sequence[BaseMessage], add_messages]

Step 2: 定义工具

@tool

def search_weather(city: str) -> str:

"""查询指定城市今天的天气。参数 city 为城市名称。"""

weather_db = {

"北京": "北京今天晴，温度 15°C ~ 28°C，湿度 35%，风力 2-3级",

"上海": "上海今天多云转阴，温度 22°C ~ 31°C，湿度 70%，风力 3-4级",

}

return weather_db.get(city, f"未找到{city}的天气数据")

@tool

def search_air_quality(city: str) -> str:

"""查询指定城市的空气质量指数(AQI)和PM2.5。参数 city 为城市名称。"""

aqi_db = {

"北京": "北京 AQI 68，PM2.5 35μg/m³，PM10 78μg/m³，空气质量：良",

"上海": "上海 AQI 45，PM2.5 22μg/m³，PM10 52μg/m³，空气质量：优",

}

return aqi_db.get(city, f"未找到{city}的空气质量数据")

tools = [search_weather, search_air_quality]

tools_by_name = {t.name: t for t in tools}

Step 3: 绑定工具到模型

model = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)

Step 4: 定义图节点

def call_model(state: AgentState) -> AgentState:

response = model.invoke(state["messages"])

return {"messages": [response]}

def call_tool(state: AgentState) -> AgentState:

last_message = state["messages"][-1]

tool_messages = []

for tool_call in last_message.tool_calls:

tool_func = tools_by_name[tool_call["name"]]

result = tool_func.invoke(tool_call["args"])

tool_messages.append(

ToolMessage(content=str(result), tool_call_id=tool_call["id"])

)

return {"messages": tool_messages}

def should_continue(state: AgentState) -> Literal["call_tool", END]:

last_message = state["messages"][-1]

if hasattr(last_message, "tool_calls") and last_message.tool_calls:

return "call_tool"

return END

Step 5: 构建并编译图

workflow = StateGraph(AgentState)

workflow.add_node("agent", call_model)

workflow.add_node("call_tool", call_tool)

workflow.set_entry_point("agent")

workflow.add_conditional_edges("agent", should_continue, {

"call_tool": "call_tool",

END: END,

})

workflow.add_edge("call_tool", "agent")

agent = workflow.compile(checkpointer=MemorySaver())

运行Agent

config = {"configurable": {"thread_id": "demo-1"}}

task = HumanMessage(content="北京今天天气怎么样？空气质量如何？适合户外运动吗？")

print("=== ReAct Agent 执行轨迹 ===\n")

for event in agent.stream({"messages": [task]}, config=config):

for node_name, node_output in event.items():

if node_name == "agent":

msg = node_output["messages"][-1]

if hasattr(msg, "tool_calls") and msg.tool_calls:

for tc in msg.tool_calls:

print(f" 🔧 调用工具: {tc['name']}({tc['args']})")

elif hasattr(msg, "content") and msg.content:

print(f" 💬 模型回复: {msg.content[:150]}...")

elif node_name == "call_tool":

for m in node_output["messages"]:

print(f" 📊 工具返回: {m.content[:120]}...")

💡 关键要点：

- bind_tools 将函数签名注入 system prompt，模型自动学会何时调用哪个函数

- 图的循环逻辑：agent → 有 tool_calls → call_tool → agent → 重复，直到无调用

- add_messages 是 reducer 函数，自动处理消息追加而非覆盖

🛠️ 实战经验：生产环境至少还要加四件事——① 工具调用失败的错误重试；② 流式输出（用 .astream_events() 逐字吐结果）；③ 单个工具的超时控制；④ max_iterations 硬上限防死循环。LangGraph 的 checkpointer 支持断点恢复——Agent 崩溃后可以从上次状态继续，这对长任务场景至关重要。

5. 常见误区

❌ 误区1：Agent = 加了工具调用的复杂 Prompt Chain

Chain 的流程是确定的——A→B→C，每一步写死。Agent 的流程是动态的——模型自己决定调什么工具、调几次、什么顺序、什么时候停。这个区别在工程层面含义巨大：Chain 可以离线调优，Agent 的行为高度依赖模型能力和运行时状态，debug 要难得多。把 Chain 当 Agent 上线，第二周就可能翻车——模型在某些步骤会跳过或调用不在计划内的工具。

❌ 误区2：MCP 只能给 Anthropic 的模型用

MCP 是 模型无关 的开放协议，使用 JSON-RPC 2.0，任何支持 HTTP 通信的 LLM 都能集成。OpenAI 的 Assistants API、Google Gemini、LangChain/LlamaIndex 都已经支持 MCP。2025 年底 MCP 捐献 Linux 基金会后，厂商中立属性更加明确。

❌ 误区3：Agent 越复杂越好——Multi-Agent 一定比单 Agent 强

Multi-Agent 引入的问题比解决的还多——通信开销、状态同步、不一致放大。我们做过 A/B 测试：同一个客服任务，GPT-4o 单 Agent 跑 ReAct 准确率 87%，三个 GPT-4o-mini 做 Multi-Agent 准确率 82%，token 消耗却是前者的 2.3 倍。不是说 Multi-Agent 没用——在需要天然专业分工的场景（如代码生成+安全审查+测试）它确实更好——但别把它当默认方案。

6. 练习与思考

练习1：基础理解

下面关于 ReAct 和 RAG 的说法，哪个是正确的？

A. ReAct 就是 RAG 加上工具调用

B. RAG 是 Agent 的一个子组件，Agent 可以调用 RAG 作为工具

C. Agent 一定能替代 RAG

D. ReAct 循环中模型不需要 Observation 就能继续推理

<details>

<summary>查看答案与解析</summary>

正确答案：B

解析：RAG（检索增强生成）本质上是一个"检索+生成"的流程，可以封装成工具让 Agent 调用——Agent 判断需要查资料时调 RAG 工具检索，拿到结果后再推理回答。A 错误，ReAct 的模式维度比 RAG 更广；C 错误，两者解决不同问题，不是替代关系；D 错误，Observation 是 ReAct 循环的必要环节。

</details>

练习2：架构设计

你正在为一家电商公司设计售后客服 Agent 系统。需求包括：查询订单状态、发起退款、回答退换货政策、升级到人工客服。

<details>

<summary>查看答案与解析</summary>

HITL 建议：发起退款必须设 HITL——不可逆的"写操作"，涉及资金变动，出错后果严重。查询订单和回答政策不需要 HITL——前者是安全读操作，后者是只读检索。升级到人工客服本质就是 HITL 的一种实现。

架构选择：建议用单 Agent（ReAct）+ 工具集。售后客服任务虽然类型多，但逻辑上是串行的——查政策→查订单→判断→退款。一个强模型配合好 prompt 能完全胜任。如果系统还包括"欺诈检测"这类需要独立判断的模块，可考虑给它配独立 Agent。Multi-Agent 在这个场景会引入不必要的编排复杂度。

</details>

练习3：拓展思考

如果让你从零设计一个 Agent 框架，ReAct、Plan-Execute、Multi-Agent、HITL 四种模式你会如何分层组合？画出你的架构分层图并说明每一层的职责。

<details>

<summary>查看思路引导</summary>

思考方向：通常的架构是——最底层是 ReAct 作为基础执行循环；上一层是 Plan-Execute 作为任务编排层，将用户指令拆解为子任务，每个子任务由一个 ReAct 实例执行；再上层是 Multi-Agent 的调度层（可选，仅在需要专业分工时引入）；贯穿所有层的是 HITL 安全拦截层，拦截所有"写操作"。这样设计的好处是：简单任务可以只用 ReAct 一层，复杂任务层层叠加，每一层独立演进和替换。

</details>

延伸阅读

---

💡 核心收获：

- RAG 到 Agent 不是"升级替代"而是能力叠加——RAG 管"记忆"，Agent 管"行动"，生产系统中两者几乎总是同时出现

- ReAct 是 Agent 的心跳节拍，Plan-Execute 是复杂任务的导航图，Multi-Agent 是万不得已的分工方案，HITL 是生产环境的安全闸

- MCP 和 A2A 不是竞争关系——MCP 管"Agent 怎么用工具"（纵向），A2A 管"Agent 之间怎么说话"（横向），两者互补共存