这两年做大模型应用只要沾到“私有数据”三个字基本绕不开 RAG 这套玩法。不管你是给公司搭内部知识库助手还是做垂直行业的问答机器人第一步想到的几乎都是检索增强生成先让模型去资料库里翻答案再基于翻到的内容组织语言输出。这套路我已经用了很多次组件换来换去最后发现把检索能力统一封装成 MCP 工具整个链路才真正顺起来。这篇教程就按我实际的搭建过程来写从零开始讲清楚基于 MCP 的 RAG 系统怎么落地包括架构怎么拆、代码怎么写、参数怎么调、坑在哪里尽量一次讲透。这篇文章适合两类朋友一类是刚开始接触 RAG想找一个干净、可复现的工程起点另一类是已经写过几个 RAG Demo但想把检索环节标准化方便多个 Agent 或应用复用。你需要有一点 Python 基础懂一点异步编程更好不过不懂也没关系代码我会拆开讲。1. 搞懂两个核心概念RAG 和 MCP 分别解决什么问题1.1 RAG让模型从“闭卷考试”变成“开卷考试”大语言模型本身有个天然短板它训练完的那一刻知识就冻结了之后你公司内部的新文档、新制度、产品新功能它一概不知道。让模型去回答这些私域问题常见方案有三个直接塞进上下文、微调、RAG。直接塞进上下文最简单但上下文窗口有限几万字的文档塞进去很快就超限而且你没法保证模型精准定位到关键段落微调成本高、周期长还要准备高质量训练数据对大多数场景来说性价比太低。RAG 的思路就聪明在不改变模型本身而是把“找答案”这个动作外部化——先通过检索从知识库里捞出来和问题最相关的片段再把片段和问题一起交给模型让模型基于这些片段来回答。打个比方闭卷考试靠脑子硬记开卷考试先翻书再作答。RAG 就是把闭卷改成开卷模型不需要记住你文档里的每一个细节只要会“翻书”就行。这样做的好处很明显答案有出处、可追溯新知识随时往知识库里加不需要重新训练模型。1.2 MCP把 AI 应用的数据连接方式统一成标准协议MCP 全称是 Model Context Protocol模型上下文协议。它解决的问题是AI 应用和外部数据源、工具之间的连接方式太碎片化。以前每接一个数据源就得写一套专用集成代码换个项目全部重来。MCP 的做法是定义一个标准协议把数据源和工具包装成统一的 ServerAI 客户端只要支持这个协议就能像插 USB 一样即插即用。为什么这个话题今天突然这么重要因为 AI 应用已经不是“单模型”逻辑了Agent 要读文件、查数据库、调 API、操作浏览器每一样都是不同接口、不同认证方式。MCP 出现之后这些能力被标准化成 Tool 或 Resource客户端通过协议调用配置一下就能用。你可以把 MCP 理解为 AI 世界的“标准插座”后端能力做成了各种插头只要规格统一插上就有电。1.3 为什么要把 RAG 和 MCP 结合起来单个 Demo 里RAG 确实不需要 MCP自己写个search()函数就够了。但一旦你开始做真实产品情况就变了你不只一个应用可能有一个客服机器人、一个内部问答助手、一个数据分析 Agent它们都需要访问同一套知识库。这时候每个应用各自实现一套检索逻辑代码重复、维护成本高、接口还不统一。把 RAG 的检索能力封装成 MCP Server等于把“检索知识库”这件事做成了一个标准服务。任何 MCP 客户端都能直接调用不用关心底层用的是什么向量库、什么 Embedding 模型、什么切分策略。新同学接手时只需要知道“有一个 search 工具传一个问题进去返回相关片段”心智负担小很多。另外MCP 生态里已经有很多现成 Server比如文件系统、数据库、浏览器工具。你把 RAG 检索作为其中一个工具接入Agent 就能在回答问题时同时查文档、查数据库、查网页组合出更复杂的能力。这才是实际产品里真正需要的形态。所以这篇教程选择直接用 MCP 作为 RAG 的包装层而不是写一个只能在自己项目里跑通的死代码。2. 系统架构设计与技术选型2.1 整体架构拆解整个系统的运行链路是这样的用户提问 → MCP 客户端调用search_documents工具 → MCP Server 接收查询 → 生成 Query 向量 → 在向量库中检索 Top-K 片段 → 返回候选内容及相似度分数 → MCP 客户端把片段和问题拼成 Prompt → 调用大模型 → 生成最终回答。这里关键点在于MCP Server 承担的是“检索”职责而“生成”放在 MCP 客户端一侧。为什么这么拆因为 MCP 工具定位是“能力暴露”它不应该绑定某一个模型或某一套 Prompt 模板。检索结果给谁用、怎么用是调用方自己的事。后续你想换模型、换 Prompt 风格完全不需要动 Server 端。组件层面的分工大概是这样组件职责关键技术MCP Server封装检索与知识管理能力fastmcp向量数据库存储文档向量、执行相似度检索ChromaEmbedding 模型将文本转为向量开源中文向量模型MCP Client调用工具、编排对话流程fastmcp ClientLLM 推理接口基于检索上下文生成回答OpenAI 兼容接口2.2 技术选型与理由先说明一下选型原则我的目标是“能让读者在本地一台机器上跑通、代码尽量少、不依赖重型服务”所以全部组件都选了轻量方案。向量库选了 Chroma不是因为它性能最好而是因为它在“单机跑通”这件事上体验最顺。安装完直接PersistentClient落盘重启数据不丢不需要额外启服务端进程。FAISS 性能也很强但没有自带持久化管理需要自己维护索引文件对于从零起步的教程来说多一层负担。等以后数据量真的大了再迁移到 Milvus 之类的专用服务也不晚Chroma 里的数据可以导出来。MCP 框架选了 fastmcp它是一个 Python 实现的 MCP SDKAPI 非常简洁一个装饰器就能暴露一个工具省去手工处理 JSON-RPC 协议细节的麻烦。需要说明的是现在的 MCP 协议版本更新得很快fastmcp 也在跟着迭代建议安装时不要锁太老的版本。Embedding 模型我选的是开源中文向量模型 bge-small-zh-v1.5维度 512中文效果在同等体量模型里表现不错而且本地推理不需要联网、不需要额外付费。如果你的语料偏英文技术文档可以换成 bge-base-en 或者其它推理资源允许的模型代码基本不用改只要换模型名重新跑一次向量化就行。LLM 接口这边我没有绑定任何特定品牌的 SDK而是直接用 HTTP 调用 OpenAI 兼容的/v1/chat/completions接口。现在绝大多数模型服务商都提供这个兼容协议你只需要配置 Base URL、API Key 和模型名代码本身是通用的。如果你完全在本地跑模型可以用支持 OpenAI 兼容协议的推理框架起一个本地服务Base URL 指向 localhost 即可。3. 环境准备与依赖安装3.1 环境要求与依赖清单建议使用 Python 3.10 及以上版本。我实际测试时用的 3.11全程没有碰到 Python 版本相关的问题。操作系统方面Windows、macOS、Linux 都能跑只是创建环境时的命令略有差异。依赖项并不多核心就这几个fastmcp chromadb sentence-transformers requests安装命令pip install fastmcp chromadb sentence-transformers requestssentence-transformers会自动拉取 PyTorch如果你的机器没有 GPU它会用 CPU 推理bge-small 这种体量的模型用 CPU 也跑得动就是向量化大批量文档时稍微慢一点。GPU 机器能明显快一些但教程案例的数据量不大CPU 完全够用。3.2 项目目录结构我习惯把工程按职责拆成几个文件方便后续维护。你可以参考这个结构mcp-rag-tutorial/ ├── build_index.py # 构建知识库向量索引 ├── server.py # MCP Server暴露检索与添加文档工具 ├── client.py # MCP Client调用工具并完成问答测试 ├── docs/ # 放原始知识文档纯文本格式即可 └── chroma_db/ # 向量库持久化目录运行时自动生成这里有一个容易忽略的点server.py启动时需要能访问到chroma_db目录所以启动命令的工作目录必须和构建索引时保持一致否则会新建一个空库导致检索不到任何内容。我最初就吃过这个亏后面在项目根目录启动就再没出过问题。4. 知识库预处理切分、向量化、存储4.1 文档加载与切分参数RAG 效果好不好第一步就看切分。切分太粗每个片段里掺了太多无关内容检索出来的 Top-K 可能只有一句有用切分太细上下文碎片化模型理解起来费劲。我一般用的配置是chunk_size 500字符overlap 80字符这个配置对中文技术文档、制度文件、产品手册都适用。为什么需要 overlap因为句子和语义的边界经常被硬生生切断比如“该功能仅对管理员开放”这句话可能被从中间劈开前半段在 chunk A后半段在 chunk B。加了重叠之后即使关键句被切在分界线上相邻 chunk 里也能找到完整内容检索命中率会明显提升。我这里的切分逻辑没有直接用第三方切分器而是写了一个简单的递归切分函数方便你理解内部机制import re def split_text(text: str, chunk_size: int 500, overlap: int 80): # 先按段落拆 paragraphs [p.strip() for p in re.split(r\n\s*\n, text) if p.strip()] chunks [] current for para in paragraphs: if len(current) len(para) 1 chunk_size: current \n para else: if current: chunks.append(current) current para # 对超长段落按句切 if current: chunks.append(current) final_chunks [] for chunk in chunks: if len(chunk) chunk_size: final_chunks.append(chunk) else: sentences re.split(r(?[。]), chunk) buf for sent in sentences: if len(buf) len(sent) chunk_size: buf sent else: if buf: final_chunks.append(buf) buf sent if buf: final_chunks.append(buf) # 加上 overlap overlapped [] for i, chunk in enumerate(final_chunks): if i 0: overlapped.append(chunk) else: pre final_chunks[i - 1][-overlap:] overlapped.append(pre chunk) return overlapped这个函数是教学版的简化实现真实项目中如果文档有明确标题结构建议优先按标题层级切分先保证语义完整性再考虑长度限制。4.2 Embedding 模型的选择与加载向量化是 RAG 的核心环节。Embedding 模型将一段文本映射为一个高维向量语义相近的文本在向量空间里的距离也更近。检索时把用户的提问向量化和库里的文本向量算相似度找最近的几个就是候选答案。我用的是sentence-transformers库加载开源中文模型from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-small-zh-v1.5)第一次运行会自动下载模型权重需要保持网络畅通。如果网络不稳可以手动下载后放到本地目录再加载。加载 Embedding 模型有一个非常影响效果的隐藏细节模型参数要不要冻结、有没有针对中文语料做领域适配。通用模型处理法律、医疗、金融这些垂直领域术语时效果会打折后续如果想提升效果可以考虑用领域微调过的专用 Embedding 模型或者在本地小规模语料上做增量训练但这是后话了。4.3 向量索引构建与保存构建索引的完整脚本build_index.pyimport os import chromadb from sentence_transformers import SentenceTransformer from split_text import split_text model SentenceTransformer(BAAI/bge-small-zh-v1.5) client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection( nameknowledge_base, metadata{hnsw:space: cosine} ) docs_dir ./docs all_chunks [] all_ids [] all_metadatas [] for idx, filename in enumerate(os.listdir(docs_dir)): filepath os.path.join(docs_dir, filename) if not filename.endswith(.txt): continue with open(filepath, r, encodingutf-8) as f: content f.read() chunks split_text(content) for cid, chunk in enumerate(chunks): uid f{filename}_{cid} all_chunks.append(chunk) all_ids.append(uid) all_metadatas.append({source: filename}) embeddings model.encode(all_chunks).tolist() collection.add( idsall_ids, documentsall_chunks, embeddingsembeddings, metadatasall_metadatas ) print(fknowledge base built: {len(all_chunks)} chunks)这里有几个设计决定需要解释一下。一是hnsw:space设为cosine因为开源 bge 模型官方建议匹配余弦相似度而且余弦距离对文本长度不那么敏感更符合语义检索场景。二是先批量 encode 再一次性写入向量库比逐条 encode 逐条 add 快很多因为模型推理本身有批量加速Batch Size 默认对 CPU 和 GPU 都更友好。index 构建好之后接下来看一下向量库的“冰山下面那一块”检索用的相似度逻辑我们用的欧氏距离和余弦距离的换算关系cosine distance 1 - cosine similarity所以在查询结果里距离越小代表越相似可视化时你可以用1 - distance还原相似度分数这样读者更容易理解。5. 实现 MCP Server把检索能力封装成标准工具5.1 用 fastmcp 快速搭建服务端server.py的核心代码from fastmcp import FastMCP import chromadb from sentence_transformers import SentenceTransformer mcp FastMCP(rag-server) _model SentenceTransformer(BAAI/bge-small-zh-v1.5) _chroma_client chromadb.PersistentClient(path./chroma_db) _collection _chroma_client.get_or_create_collection( nameknowledge_base, metadata{hnsw:space: cosine} ) mcp.tool def search_documents(query: str, top_k: int 5) - list[dict]: 根据用户问题检索知识库中最相关的文档片段。 Args: query: 用户的自然语言问题。 top_k: 返回相关片段数量默认 5。 q_embedding _model.encode([query]).tolist() res _collection.query( query_embeddingsq_embedding, n_resultstop_k, include[documents, distances, metadatas] ) results [] for doc, dist, meta in zip( res[documents][0], res[distances][0], res[metadatas][0] ): score round(1 - dist, 4) # 余弦距离转相似度 results.append({ text: doc, score: score, source: meta.get(source, unknown) }) return results mcp.tool def add_document(text: str, source: str manual) - str: 向知识库添加一条新文档用于增量更新知识库。 uid fmanual_{abs(hash(text))} embedding _model.encode([text]).tolist() _collection.add( ids[uid], documents[text], embeddingsembedding, metadatas[{source: source}] ) return fadded: {uid}注意每个工具函数的 docstring 不能省。MCP 协议会把函数签名和 docstring 发给支持模型推理的客户端LLM 根据这些描述来决定是否调用工具、传什么参数。docstring 写清楚了Agent 才能正确使用工具。5.2 核心工具的设计思路search_documents是主工具负责整个 RAG 的检索环节。返回结构里除了文本本身我还带了score和source这两个字段在实际项目中非常有用。score用来判断检索置信度如果 Top-1 的相似度连 0.5 都不到你可能要告诉用户“知识库中没有找到相关答案”而不是硬堆给模型。source则用于后续溯源回答附上引用来源用户信任度高不少。add_document这个工具是故意加的它让知识库具备了运行时增量写入能力。实际场景中新文档不断产生不可能每次都离线重建索引。通过 MCP 工具暴露添加接口之后任何客户端都可以在对话过程中把新知识写入知识库实现“边聊边学”的效果。为什么把检索和写入都做成 MCP 工具而不是 REST APIREST API 需要额外鉴权、定义接口文档、处理各种 Client 接入问题MCP 工具天然在协议层解决了这些问题客户端只需要列出可用工具就能自动发现能力。这也是 MCP 标准化带来的最大红利。5.3 启动 Server在项目根目录执行python server.py正常情况下 fastmcp 会启动并等待客户端连接。默认传输方式下Client 会以子进程方式拉起 Server所以不需要手动先启动 Server直接跑客户端即可。如果你是手动调试也可以用mcp dev server.py这类命令启动开发模式但教程里直接让 Client 管理子进程更省心。6. 实现 MCP Client 与 RAG 问答链路6.1 客户端连接client.py负责连接 MCP Server、调用工具并把检索结果交给 LLM。核心代码import asyncio from fastmcp import Client async def main(): async with Client(python server.py) as client: tools await client.list_tools() print(mcp tools:, [t.name for t in tools]) result await client.call_tool( search_documents, {query: 什么是MCP协议, top_k: 3} ) print(search result:) for item in result: print(item) if __name__ __main__: asyncio.run(main())这里Client(python server.py)是 fastmcp 的 Stdio 传输方式客户端自动拉起 Server 子进程通过标准输入输出进行 JSON-RPC 通信。好处是不需要管理网络端口本地调试非常方便。等你要部署到远程环境时再切换成 Streamable HTTP 传输即可。6.2 构建增强 Prompt检索到相关内容之后不能直接把一堆片段丢给模型它不知道哪些重要哪些不重要。我常用的 Prompt 模板是这样的SYSTEM_PROMPT 你是一个知识库问答助手。请根据用户提供的检索资料回答问题。 要求 1. 仅基于检索资料中的信息作答不要编造事实。 2. 如果资料不足请直接说明没有找到相关信息。 3. 回答时标注引用来源。 def build_prompt(question: str, contexts: list[str]) - str: context_block \n\n---\n\n.join(contexts) return f以下是知识库中检索到的相关资料 {context_block} 用户问题{question} 请基于以上资料给出答案。这段 Prompt 有三点值得注意一是强约束“不要编造”尽可能压制 LLM 的幻觉倾向二是给模型明确的资料上下文边界三是要求标注来源逼着模型去用上下文而不是自己脑补。6.3 调用 LLM 生成答案LLM 调用我用纯 HTTP 写方便你对接任意 OpenAI 兼容服务import os import requests API_BASE os.getenv(LLM_API_BASE, http://localhost:8000/v1) API_KEY os.getenv(LLM_API_KEY, local) MODEL_NAME os.getenv(LLM_MODEL, qwen2.5-7b-instruct) def call_llm(messages: list[dict]) - str: resp requests.post( f{API_BASE}/chat/completions, headers{Authorization: fBearer {API_KEY}}, json{model: MODEL_NAME, messages: messages, temperature: 0.2}, timeout120, ) resp.raise_for_status() return resp.json()[choices][0][message][content]把之前的搜索和生成串起来async def ask(question: str): async with Client(python server.py) as client: search_result await client.call_tool( search_documents, {query: question, top_k: 5} ) contexts [r[text] for r in search_result] request_messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: build_prompt(question, contexts)}, ] answer call_llm(request_messages) return answertemperature设到 0.2 是刻意为之问答场景要的是稳定、忠实于资料而不是创造性发挥。如果你做的是头脑风暴类应用温度可以拉高但 RAG 问答场景低温度是默认选择。7. 完整测试与调优7.1 端到端测试现在把整个链路跑一遍确认所有环节正常。先放一份测试文档到docs/内容可以写一小段介绍文字比如某产品的使用规范。然后依次执行python build_index.py python client.py如果一切正常你应该看到工具列表里有search_documents和add_document检索返回的片段和查询语义基本匹配。然后再跑一个带 LLM 的问答脚本看最终答案是否正确引用了文档里的内容。我测试时习惯先不加 LLM直接看检索结果。这一步能有效区分是检索的问题还是生成的问题。如果检索结果本身就不相关再怎么调 Prompt 也没用如果检索结果相关但答案胡扯那才是生成环节的问题。7.2 核心参数调优RAG 系统最值得调的四个参数是chunk_size、overlap、top_k和相似度阈值。chunk_size影响的是粒度片段太长干扰信息多片段太短语义不完整。一个可以参考的判断方法是把检索到的 Top-1 片段拿出来人工读一遍如果它能完整回答你的测试问题说明粒度合适如果只答了一半说明切得太碎需要调大chunk_size如果读起来冗长但关键信息夹在中间说明调小一点更精准。top_k影响的是最终 Prompt 的上下文量。不是越大越好因为 Token 有限塞太多无关片段反而会误导模型。我一般先用top_k 5起步然后观察相似度分数的衰减曲线如果第 3 个片段的分数已经低于 0.5那top_k设 3 就合适了如果前 5 个分数都不错可以加到 8。相似度阈值是个容易被忽略但很实用的参数。建议在search_documents里加一道过滤MIN_SCORE 0.6 filtered [r for r in results if r[score] MIN_SCORE]如果过滤后一个结果都没有说明知识库里大概率没有相关内容这时候直接让模型说“未找到”比硬答效果好得多也体面得多。我用一个测试问题示例调参过程问题“某某功能如何开启”最开始chunk_size1000、top_k3检索出的第一个片段包含了大段背景介绍关键操作步骤被淹没把chunk_size改成 500 后Top-1 恰好就是包含操作步骤的段落再把overlap从 0 调成 80边界句子的召回明显改善。这三个参数是一组最好一起试验而不是单独调某一个。8. 踩坑实录常见问题与排查方法8.1 常见问题速查表现象可能原因解决方法客户端连接 Server 超时启动路径不对Server 子进程崩溃检查Client(python server.py)路径先单独运行python server.py看报错检索结果为空向量库为空或集合名不一致确认build_index.py和server.py中collection名称、chroma_db路径一致检索结果全不相关切分粒度太大或太小Embedding 模型与语料不匹配先人工检查切分质量再尝试换领域适配的模型分数全部很低HNSW 空间配置错误或用了欧氏距离确认metadata{hnsw:space: cosine}用1 - distance还原相似度LLM 报 401/404API Base 或模型名配置错误检查环境变量先用 curl 直接调接口确认可用性重复建库后旧数据残留Chroma 持久化目录复用新增集合名或用collection.delete()清理旧数据回答不引用资料Prompt 里没有强调引用来源在 Prompt 中明确要求“标注引用来源”8.2 几个容易忽略的细节第一MCP 工具 docstring 一定要写清楚参数含义。我带一个新手尝试接入 Agent 框架时工具描述写得不清楚LLM 总是把top_k传成字符串类型导致报错。后来把 docstring 写成“返回相关片段数量默认 5”问题立刻消失。第二Embedding 模型和向量库的空间配置要匹配。Chroma 的默认 distance 可能是 l2如果你用 bge 且不明确设置cosine结果排序逻辑会和预期不一致虽然也能用但相似度分数的含义就变味了。第三不要在build_index.py里同时跑太多文档。第一次把几百个文档一次性塞进去时Python 的 GIL 和多线程加锁问题会导致 encode 阶段内存暴涨。分批处理是更稳妥的做法比如每 100 个文档做一次批量 encode 和 add再加进度输出方便观察耗时。第四MCP 协议版本兼容性问题。fastmcp 迭代速度很快如果后续升级了包Client和call_tool的行为可能有细微变化。我的建议是开发阶段用一个固定版本锁在 requirements.txt 里不要盲目追新。9. 一点个人心得把 RAG 装进 MCP 容器之后我最大的感受是这套架构把“能力”和“使用方式”解耦得非常干净。以前写 RAG代码大多纠缠在一个项目里换一个场景就要复制粘贴改一堆东西。现在能力沉淀在 Server 端新项目来了接 MCP 客户端配置一下就能用我自己的效率提升非常明显。最后再分享一个小技巧调试 MCP 工具时强烈建议先单独调工具、再调完整链路。你可以先用client.call_tool只跑检索不接 LLM确认检索质量过关了再往上叠加生成环节。很多人一上来就端到端调试结果出了问题不知道是检索的问题还是模型的问题排查半天。按这个顺序来能把变量一个一个隔离掉定位速度会快很多。如果后续你想在这个系统上继续扩展有几个方向可以参考在检索环节加入 Reranker 做二次精排效果提升非常明显把 MCP Server 部署成远程 HTTP 服务支持多客户端接入对 Embedding 模型做领域微调解决垂直术语的召回问题。这条路的扩展空间还很大祝你在上面玩得开心。