当前位置：首页 > news >正文

RAG 与 MCP：每位 AI 开发人员都必须了解的架构差异

news 2026/4/15 17:20:01

我第一次听到两个工程师争论 RAG 和 MCP 的时候，还真以为他们在讨论宝可梦的进化。其中一个不停地坚持说：“它们本质上是一样的。”另一个则自信地点头，好像刚解决了分布式系统的问题。与此同时，他们开发的生产Agent却像个糊涂蛋一样呆在角落里，它觉得自己能背诵公司里所有的政策文件，却查不出五分钟前哪个流水线出了问题。真是经典。

这时我才恍然大悟：大多数开发者把 RAG 和 MCP 当作可以互换的流行语，而实际上它们解决的是完全不同的问题。

RAG 是你的智能体的记忆：它记住了你组织曾经编写过的所有内容。MCP 是它的双手：它负责在现实世界中实际操作。如果把它们混在一起，你最终会得到一个书本知识丰富但在实际系统中完全无用的 AI。这就像雇佣了一个能背诵你整个维基百科页面的人，但当你让他打开浏览器标签页时，他却会立刻卡壳。

本文旨在解决这个问题。读完本文后，你不仅会熟悉相关概念，还会掌握相应的思维模型、理论知识以及可运行的代码示例，从而真正理解并运用到生产环境中。

简单的区分

RAG = 阅读 → 从你的LLM培训中未涉及的文档中提取知识；MCP = 执行 → 对实时系统和实时数据执行操作

记住这一点。其他一切都由此而来。

第一部分：RAG——赋予你的LLM长期记忆

什么是 RAG？

检索增强生成 (RAG) 是一种模式，其中您可以使用在查询时从您控制的知识库中检索的相关内容来增强 LLM 的提示。

LLM的学习进度停滞在培训截止日期。他们对以下方面一无所知：

您的内部文档
上个季度的架构决策
团队的运行手册或服务级别协议
任何在截止日期之后写的东西

RAG算法解决了这个问题。它预先将文档索引到向量存储库中，当用户提出问题时，系统会提取语义最相关的文本块，并将其注入到LLM的上下文窗口中。这样，模型无需重新训练即可获得“即时知识”。

实际上，这意味着如果有人问“我们的管道重试服务级别协议 (SLA) 是什么？”，模型不会猜测，而是会检索准确的策略文本，并用基于事实的答案来回答。RAG 是 AI 代理的记忆层。它告诉模型你的组织知道什么，而不是它应该做什么。

RAG 流程

何时使用 RAG？

当你的知识是静态的或半静态的时，RAG 是完美的——例如文档、操作手册、合同、白皮书、入职指南。

第二部分：MCP——运用您的LLM技能

什么是MCP？

模型上下文协议 (MCP) 是一个开放标准（由 Anthropic 引入），它定义了 LLM调用外部工具、API 和实时系统的结构化方式。

把它想象成带有形式化规范的函数调用。它不会像法学硕士那样凭空臆想出关于实时数据的答案，而是会说：

我需要查一下。让我调用一下合适的工具。

MCP 工具可以：

查询实时数据库
触发 Databricks 工作流
从Kafka主题中阅读
调用 REST API
将记录写入记录系统

换句话说，MCP 将你的 LLM 从只会空谈的模型转变为会行动的模型。它使模型能够通过类型化的模式、防护措施和可预测的行为，对真实系统进行受控访问。

例如，如果用户问：“昨晚的管道运行失败了吗？”，模型不会猜测，而是调用你的get_pipeline_status工具，获取实时结果，并返回真实数据。

MCP是AI代理的行动层。它告诉模型系统当前正在做什么，而不是文档中描述的是什么。

MCP流程

以下是扩展后的“何时使用 MCP”表格，以可视化决策图的形式呈现：

第三部分：真正的力量——两者结合使用

真实用户的问题并不简单。他们常常把知识和操作混杂在一句话里：

“我们针对管道故障的服务水平协议 (SLA) 是什么？昨晚的销售数据导入作业是否违反了该协议？”

这个问题包含两个部分：

“我们的服务级别协议 (SLA) 是什么？” → 从静态运行手册中检索 →RAG

“昨晚是否发生违规？” → 查询实时作业监控 API →MCP

纯粹的 RAG 系统无法回答第二个问题。它无法访问实时系统。纯粹的 MCP 系统除非您将其硬编码到系统中，否则无法了解您的 SLA。换句话说，纯粹的 MCP 系统无法回答第一个问题，除非您将每项策略都硬编码到工具中，否则它没有长期记忆功能。它们结合起来，既能赋予您的代理记忆功能，又能赋予其自主权。

→ RAG 告诉模型你的组织了解哪些信息。→ MCP 告诉模型你的系统当前正在执行哪些操作。→ 现代 AI 代理需要这两者才能在实际工作流程中智能地运行。

第四部分：代码——自己动手构建

项目：数据管道支持代理

一个模块化代理，可以回答有关管道策略（RAG）的问题，并检查实时作业运行状态（MCP 风格的工具调用）。无需 Databricks 或 Airflow——所有实时数据均为模拟数据。

项目结构

rag_vs_mcp_demo/ ├── agent/ │ ├── llm_router.py # 基于关键词的 RAG/MCP/混合分类器 │ ├── openai_agent.py # 主代理：RAG + OpenAI 工具调用 │ ├── openai_client.py # OpenAI 客户端 + 工具模式 │ └── orchestrator.py # 编排器（无需 LLM） ├── rag/ │ ├── embedder.py # HuggingFace 嵌入封装 │ ├── retriever.py # 数据块 → 嵌入 → FAISS → 检索 │ └── vector_store.py # FAISS 封装 ├── tools/ │ ├── fake_api.py # 模拟延迟 + 故障 │ ├── pipeline_status.py # 模拟实时管道状态数据 │ ├── router.py # 工具调度器 │ └── row_count.py # 模拟行计数工具 ├── data/ │ ├── runbook_sla.md │ ├── dataquality_standards.md │ └── oncall_responsibilities.md ├── examples/ │ └── demo_query.py ├── .env.example └── requirements.txt

设置

# 1. 创建并激活虚拟环境 python -m venv venv source venv/bin/activate # macOS/Linux # venv\Scripts\activate # Windows # 2. 安装依赖项 pip install -r requirements.txt # 3. 设置您的 OpenAI API 密钥 cp .env.example .env # 打开 .env 文件并粘贴您的密钥：OPENAI_API_KEY=sk-... 首次运行会下载all-MiniLM-L6-v2嵌入模型（约80MB）。之后会进行缓存。 其中requirements.txt包括： requirements.txt: openai> =1.30.0 anthropic> =0.25.0 langchain> =0.2.0 langchain-community> =0.2.0 faiss-cpu> =1.7.4 sentence-transformers> =2.7.0 python-dotenv> =1.0.0

步骤 1 — 将您的政策文件存储为单独的文件

该存储库不会将策略文本硬编码到脚本.md中，而是将知识存储在专用文件中data/。这样就可以轻松添加新文档而无需修改代码——只需将新.md文件放入文件夹，它就会在启动时自动加载。

数据/

├── runbook_sla.md # 管道 SLA 阈值和重试次数上限 ├── dataquality_standards.md # 空值率、重复值和模式漂移规则 └── oncall_responsibilities.md # 确认窗口和 RCA 要求

例子 - data/runbook_sla.md：

# 管道服务级别协议 (SLA) 策略所有关键数据摄取管道必须在4小时内完成。如果管道运行时间超过4小时，则必须自动触发事件。失败率阈值：总记录数的2%。

重试次数上限为3次，超过 3次将触发事件升级。

步骤 2 — 构建 RAG 流水线（模块化）

RAG 逻辑分为三个文件：embedder.py封装 HuggingFace 模型、vector_store.py封装 FAISS 并将retriever.py它们连接在一起。

rag/retriever.py— 端到端的 RAG 流水线：

rag/retriever.py— 端到端的 RAG 流水线： from langchain.text_splitter import RecursiveCharacterTextSplitter from .embedder import Embedder from .vector_store import VectorStore class RAGRetriever : """ 端到端 RAG 管道： - 分割文档 - 嵌入块 - 存储在 FAISS 中 - 检索前 k 个块 """ def __init__ ( self, chunk_size= 300 , chunk_overlap= 30 ): self.splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap ) self.embedder = Embedder() self.vectorstore = VectorStore(self.embedder.model) def index ( self, raw_text: str ): """ 分割 + 嵌入 + 存储。""" chunks = self.splitter.create_documents([raw_text]) self.vectorstore.build(chunks) def retrieve ( self, query: str ) -> str : """返回前 k 个内容块，以单个字符串的形式返回。""" docs = self.vectorstore.search(query, k= 3 ) return "\n\n" .join([doc.page_content for doc in docs])

步骤 3 — 定义 MCP 工具（实时作业监控器）

LIVE_JOB_RUNS = { "sales_ingestion" : { "run_id" : "run_20240407_001" , "status" : "FAILED" , "duration_minutes" : 312 , "failure_reason" : "超时，耗时5小时12分钟" , "records_processed" : 1_840_000 , "failure_rate_pct" : 0.8 , "retries" : 3 , "last_run" : "2024-04-07 02:15:00" }, "inventory_sync" : { "run_id" : "run_20240407_002" , "status" : "SUCCESS" , "duration_minutes" : 87 , "failure_rate_pct" : 0.1 , "records_processed" : 540_000 , "retries" : 0 , "last_run" : "2024-04-07 03:42:00" }, "customer_360" : { "run_id" : "run_20240407_003" , "status" : "RUNNING" , "duration_minutes" : 220 , "failure_rate_pct" : None , "records_processed" : 920_000 , "retries" : 1 , "last_run" : "2024-04-07 01:00:00" } } def get_pipeline_run_status ( pipeline_name: str ) -> dict : """ 伪 MCP 工具：返回实时管道状态。 在生产环境中，请替换为 Databricks / Airflow / REST API。 """ key = pipeline_name.lower().replace( " " , "_" ) return LIVE_JOB_RUNS.get(key, { "error" : f"未找到管道：{pipeline_name} " })

步骤 4 — LLM 路由器

agent/llm_router.py— 一个轻量级的关键词分类器，用于判断查询需要 RAG、MCP 还是两者都需要。这样就避免了为了路由意图而额外调用 LLM。

class LLMRouter : """ 决定查询是否需要： - RAG（知识检索） - MCP（实时工具调用） - HYBRID（两者兼有） """ def classify ( self, query: str ) -> str : q = query.lower() mcp_keywords = [ "status" , "failed" , "running" , "breach" , "row count" , "records" , "live" , "today" , "last night" , "trigger" , "restart" , "job" ] rag_keywords = [ "what is" , "explain" , "policy" , "runbook" , "sla" , "documentation" , "standard" , "definition" ] needs_mcp = any (k in q for k in mcp_keywords) needs_rag = any (k in q for k in mcp_keywords)在rag_keywords 中） if needs_mcp and needs_rag： return “hybrid” if needs_mcp： return “mcp” if needs_rag： return “rag” return “unknown”

步骤 5 — 构建代理（RAG + MCP 结合使用）

import json import os import glob from agent.openai_client import client, mcp_tools from tools.pipeline_status import get_pipeline_run_status from rag import RAGRetriever # ── 在模块加载时初始化 RAG 一次 ────────────────────────── def _load_data_docs () -> str : data_dir = os.path.join(os.path.dirname(__file__), "." , "data" ) md_files = glob.glob(os.path.join(data_dir, "*.md" )) if not md_files: raise FileNotFoundError( f"在{data_dir}中未找到 .md 文件" ) combined = [] for path in sorted (md_files): with open (path, "r" ) as f: combined.append(f.read()) print ( f"[RAG] 已加载{ len (md_files)} doc(s) from data/" ) return "\n\n" .join(combined) _rag = RAGRetriever() _rag.index(_load_data_docs()) # ── Agent ────────────────────────────────────────────────────────── def pipeline_support_agent ( user_query: str ): print ( f"\n { '=' * 60 } " ) print ( f"USER QUERY: {user_query} " ) print ( f" { '=' * 60 } \n" ) # Step 1: RAG — retrieve relevant policy context policy_context = _rag.retrieve(user_query) print ( f"[RAG] Retrieved policy context:\n {policy_context} \n" ) print ( "-" * 40 ) system_prompt = """ You作为一名数据工程支持人员， 您需要具备以下两项能力：1. 了解内部管道策略（具体内容见下文）。2 . 能够调用实时监控工具来查看实际的管道运行数据。 规则： - 如果问题需要策略/SLA 方面的知识：请使用提供的上下文。 - 如果需要实时管道状态：调用 get_pipeline_run_status 工具。- 如果需要两者兼有：同时执行这两项操作，然后综合得出清晰的答案。- 务必具体说明。如果违反了服务级别协议 (SLA)，请用数字明确说明。 """ messages = [ { "role" : "system" , "content" : system_prompt}, { "role" : "user" , "content" : ( f"策略和运行手册上下文（来自内部文档）：\n" f"---\n {policy_context} \n---\n\n" f"用户问题：{user_query} \n\n" f"回答此问题。如果您需要实时管道数据，请使用此工具。" ) } ] # 步骤 2：首次 LLM 调用 — 可能会触发工具调用 response = client.chat.completions.create( model= "gpt-4.1" , messages=messages, tools=mcp_tools, tool_choice= "auto" ) message = response.choices[ 0 ].message # 步骤 3：处理触发的工具调用 if message.tool_calls: tool_call = message.tool_calls[ 0 ] args = json.loads(tool_call.function.arguments) pipeline_name = args[ "pipeline_name" ] print ( f"[MCP] 工具调用：{tool_call.function.name} " ) print ( f"[MCP] 管道：{pipeline_name} " ) tool_result = get_pipeline_run_status(pipeline_name) print ( f"[MCP] 实时结果：{json.dumps(tool_result, indent= 2 )} \n" ) print ( "-" * 40 ) # 将工具结果反馈到最终合成 messages.append(message) messages.append({ "role" : "tool" , "tool_call_id" : tool_call. id , "content" : json.dumps(tool_result) }) final = client.chat.completions.create( model= “gpt-4.1”， messages=messages ) print ( f"[AGENT ANSWER]\n {final.choices[ 0 ].message.content} " )

# 纯 RAG 答案 — 无需工具 print ( f"[代理答案]\n {message.content} " )

第六步——运行它

始终从项目根目录运行，不要从项目内部运行examples/：

cd rag_vs_mcp_demo python -m examples.demo_query examples/demo_query.py： from agent.openai_agent import pipeline_support_agent # 查询 1：纯 RAG — 仅策略问题 pipeline_support_agent( "我们的管道故障 SLA 是多少？" ) # 查询 2：纯 MCP — 仅实时数据 pipeline_support_agent( "昨晚 sales_ingestion 作业失败了吗？" ) # 查询 3：混合 — RAG + MCP 结合 pipeline_support_agent( "我们的 SLA 是多少？sales_ingestion 是否违反了 SLA？" ) # 查询 4：MCP 行数统计工具 pipeline_support_agent( "请提供 customers 表的行数。" )

示例输出（混合查询）：

=============================================================用户查询：我们的服务级别协议 (SLA) 是什么？销售数据摄取是否违反了该协议？============================================================ [RAG] 从 data/ 加载了 3 个文档[RAG] 检索到策略上下文：所有关键数据摄取管道必须在 4 小时内完成。如果管道运行时间超过 4 小时，则必须自动触发事件。管道重试次数上限为 3 次，超过 3 次将触发事件升级。[MCP] 工具调用：get_pipeline_run_status [MCP] 管道：sales_ingestion [MCP] 实时结果：{ "status": "FAILED", "duration_minutes": 312, "failure_reason": "超时，5 小时 12 分钟", "retries": 3 } [客服回复]是的——sales_ingestion 管道昨晚违反了您的 SLA。服务级别协议 (SLA) 违规：- 策略要求在 4 小时（240 分钟）内完成- 昨晚的运行耗时 312 分钟（5 小时 12 分钟）- 这违反了 72 分钟的 SLA，足以触发自动事件。重试升级：- 管道已达到最大重试次数上限（3 次）- 根据您的策略，这将触发升级 - 不应再次自动重试。所需操作（根据您的运行手册）：1. 如果尚未创建事件单，请立即创建2. 值班工程师必须在 15 分钟内确认3. 必须在问题解决后 24 小时内提交根本原因分析。

第五部分：扩展存储库

要添加实际的管道，请将LIVE_JOB_RUNS其替换tools/pipeline_status.py为对 Databricks、Airflow 或您自己的监控系统的实时 API 调用。要添加更多知识文档，只需将任何.md文件拖放到指定位置即可data/——它们会在启动时自动加载，无需任何代码更改。