先抛一个我最近反复跟人讲的观点RAG 系统好不好用七成取决于检索那一环做得怎么样。而我这次从零搭建基于 MCP 的 RAG 系统最大的收获不是调通了一个大语言模型问答 Demo而是把“检索”从业务代码里彻底拆出来变成了一个可以通过标准协议访问的能力。本文会老老实实记录这个最小闭环的完整搭建过程包括手写 MCP 服务、分块和 embedding 的细节、完整问答链路代码以及联调时踩过的三个大坑。适合已经跑过简单 RAG、想进一步把检索服务化的开发者参考也适合想搞懂 MCP 协议到底怎么用的新人。1. 这个 Demo 想解决的痛点RAG 里的检索为什么需要 MCP1.1 一段典型的硬编码 RAG 长什么样如果只是给自己做个玩具RAG 的门槛确实不高。把一批文档切碎、向量化、丢进向量库等用户提问时再把相似片段捞出来拼进提示词一起送给大模型看起来是一个挺顺滑的闭环。但这类实现有个共性检索逻辑是直接写在应用里的。打个比方某次我给公司内部手册做问答前端调用后端接口后端先查了员工手册的向量库再调合同模板的检索函数两份资料的结果分别在不同的函数里处理。后来业务方说还想接入会议纪要和产品文档我发现自己每接一个数据源就要复制一段几乎一样的检索代码改参数、改索引名、改返回字段最后塞进一个大杂烩接口里。这和 RAG 本身关系不大纯粹是接口边界没有划好。检索这一步本应该像“查数据库”一样是个标准能力结果它和具体业务耦合在了一起。换成做 Agent 的团队感受会更明显你希望模型在需要事实支撑时自己决定去查资料可查资料的代码是服务端写死的模型只能被动拿到结果既说不上“自主”也谈不上“协作”。1.2 MCP 把检索从“函数”变成了“能力”MCP 在这里真正解决的问题是给 RAG 的检索环节定了一个统一协议。它把知识检索描述成模型可发现、可调用的工具先通过 initialize 建立会话用 tools/list 拿到当前知识库里有哪些检索能力再用 tools/call 实际执行一次查询。检索服务从此不再隶属于某一个问答 Demo而是一个独立的服务端点。这么做的好处很直接第一模型侧不再需要预置任何业务知识只要知道“有个工具能查资料”具体查什么、去哪儿查都由检索服务内部决定第二多个 Agent 可以复用同一个检索服务不用各自重写查询逻辑第三新增知识库只是给 MCP 服务增加一个工具定义不碰已有代码数据源之间天然隔离。这些价值从小 Demo 里可能看不出来一旦知识库数量超过两个、接入方超过一个团队差距会非常明显。这篇文章会带你走完整个最小闭环手写一个不依赖重型框架的最简 MCP 服务端实现向量检索的核心链路最后让大模型自主决定“要不要查、查什么”把结果带进生成阶段。代码量不大但我把所有关键决策和踩坑点都放在了对应步骤里。2. 不依赖重型框架先从零实现一个最简 MCP 服务2.1 MCP 最小协议子集initialize、tools/list、tools/callMCP 的完整规范里有很多细节但我们的 RAG 系统只需要用到三个方法理解它们比背规范重要得多。initialize 是客户端和服务端之间的握手客户端告诉服务端自己支持的协议版本服务端在返回结果里声明自己的能力集比如是否支持工具、是否支持资源。从这以后会话才正式开始。tools/list 用来获取当前可用的工具清单RAG 场景下通常就是“检索知识库”这类工具每个工具带有名字、描述和参数结构。tools/call 则是实际执行工具调用传入工具名和参数拿到结果。这三个动作可以用一个生活化的类比来解释initialize 相当于进门后互相确认身份和权限tools/list 是服务员递给你一张菜单tools/call 是你照着菜单点菜后厨做完再端上来。MCP 底层走的是 JSON-RPC 协议通信内容就是一段带 jsonrpc、id、method、params 字段的 JSON理解了这个模型你就不会被各家 SDK 绕晕。2.2 用标准库实现服务端为了把协议本身弄清楚我没有一上来就接重型依赖而是直接用 Python 标准库里的 http.server 实现了一个最简服务端。核心结构是接收 POST 请求按 method 字段分发到三个处理函数里最后把结果包成 JSON-RPC 响应返回。import json from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer TOOLS [{ name: retrieve_knowledge, description: 从知识库中检索与问题最相关的内容片段适合回答需要事实依据的问题, inputSchema: { type: object, properties: { query: {type: string, description: 用户的原始问题}, top_k: {type: integer, description: 返回几个片段, default: 3} }, required: [query] } }] def do_initialize(): return { protocolVersion: 2025-06-18, capabilities: {tools: {}}, serverInfo: {name: rag-mcp-server, version: 0.1.0} } def do_list_tools(): return {tools: TOOLS} def do_call_tool(name, arguments): if name retrieve_knowledge: # 后面几章会实现 search_knowledge result search_knowledge(arguments[query], arguments.get(top_k, 3)) return {content: [{type: text, text: result}]} raise ValueError(funknown tool: {name}) class MCPHandler(BaseHTTPRequestHandler): def do_POST(self): length int(self.headers.get(Content-Length, 0)) body json.loads(self.rfile.read(length) or b{}) try: method body.get(method) if method initialize: result do_initialize() elif method tools/list: result do_list_tools() elif method tools/call: params body.get(params, {}) result do_call_tool(params.get(name), params.get(arguments, {})) else: result {error: method not supported} except Exception as e: result {error: str(e)} response json.dumps({jsonrpc: 2.0, id: body.get(id), result: result}).encode() self.send_response(200) self.send_header(Content-Type, application/json) self.end_headers() self.wfile.write(response) if __name__ __main__: server ThreadingHTTPServer((0.0.0.0, 8000), MCPHandler) server.serve_forever()注意我用的是 ThreadingHTTPServer 而不是 HTTPServer这点非常关键。MCP 客户端可能在同一个连接里并发发送多个请求如果服务端是单线程的后到的请求会被前面的处理卡住实际联调时很容易出现假超时。2.3 用原生 HTTP 客户端调用绕过 SDK服务端写完之后客户端这边我也建议先别接 SDK直接用 HTTP 请求走一遍协议确认整个链路是通的。import json import urllib.request def mcp_call(method, paramsNone): payload json.dumps({jsonrpc: 2.0, id: 1, method: method, params: params or {}}).encode() req urllib.request.Request( http://127.0.0.1:8000/mcp, datapayload, headers{Content-Type: application/json} ) resp json.loads(urllib.request.urlopen(req).read()) return resp.get(result) print(mcp_call(initialize)) print(mcp_call(tools/list)) print(mcp_call(tools/call, {name: retrieve_knowledge, arguments: {query: 合同有效期是多久}}))调用 tools/list 时重点看一下 inputSchema它决定了模型侧怎么构造参数。很多新手在这一步会把参数名写错比如 schema 里定义的是 query调用时却传成了 question结果检索服务收到空字符串召回质量直接崩塌。协议不管你的参数名是否智能它只负责原样传递命名统一性必须靠自己在服务端校验。2.4 一个常见误区把 MCP 和 SDK 绑定在一起我见过不少项目把 MCP 客户端的功能直接嵌进业务代码里初始化逻辑、工具列表缓存、传输层全和 SDK 深度耦合。一旦想换一个 MCP 服务端实现或者从本地调试切到远程服务就要改一大堆代码。正确的做法是你的业务代码只依赖“mcp_call”这样一个薄函数所有协议细节都封装在后面这样无论底层怎么换上层 Agent 逻辑完全不受影响。这不是过度设计等接入的服务多了你就会发现薄接口是唯一能让你不崩溃的解法。3. 检索链路核心分块、嵌入、向量存储与相似度计算3.1 分块策略固定尺寸与语义边界怎么平衡MCP 服务有了但真正决定 RAG 效果的是检索质量而检索质量的第一道关是分块。我见过两种极端一种是一股脑把整篇文档塞进一个向量里结果一个长文档里只有一句话和问题相关检索时却被埋没在上千字的无关内容里另一种是切得太碎比如每 50 个字一块结果一个完整的意思被拦腰截断检索到的片段语义不完整。我常用的策略是固定长度加滑动重叠同时尽量在句号、换行这类自然边界处断块。具体到内部手册这种中文文档我习惯把块长设在 512 个字符左右重叠 64 个字符。这个配置不是拍脑袋定的而是因为 embedding 模型对过长的文本语义会稀释512 个字符既保留了一个完整段落的主干信息又不会让单个向量“装”得太多。重叠的意义在于一个关键句如果刚好被上一块的结尾和下一块的开头各截一半那么它和哪个向量都不完整匹配。加上重叠后这句话至少会完整出现在某一个块里召回的可靠性会明显提升。对于有标题结构的文档我还会在分块时把标题作为前缀拼进每一块的内容里这样“第五章 请假制度”这类实体就不会被淹没。3.2 选择 embedding 模型的思路embedding 模型的选择直接决定了向量空间里的“语义距离”是否可靠。我的建议是别盲目追求榜单上的高分模型先用你自己的领域语料做一个小样本测试准备十来组意思相近和意思相反的句子对分别计算相似度看模型能不能把它们分开。比如“请假需提前一天申请”和“事假应该提前报备”应该很近和“加班要填写审批单”应该明显远一些。如果两个完全不同语义的句子相似度都飙到 0.9 以上这个模型在你的场景里基本不能用。另外要注意向量归一化问题很多服务返回的向量本身没有归一化计算余弦相似度时影响不大但如果你要按点积近似或者做向量拼接就必须先归一化否则数值尺度不一结果全靠运气。批量 embedding 还有一个容易遗漏的点服务端通常按 batch 处理但你一次性提交的文本越多出现超时的概率越大。我习惯把一批控制在 32 条以内逐批写入向量库。写入时顺手保存一份原文和元数据后续排错会方便得多。3.3 自己写一个够用的向量检索如果你只是原型验证没必要一开始就上专门的向量数据库。我先用纯 Python 实现了一个最简单的暴力检索几千条文档的规模下性能完全可以接受。这样做的好处是能看清检索的每一步不被数据库的索引机制干扰。import math def cosine(a, b): dot sum(x * y for x, y in zip(a, b)) na math.sqrt(sum(x * x for x in a)) nb math.sqrt(sum(x * x for x in b)) return dot / (na * nb 1e-8) class VectorStore: def __init__(self): self.docs [] self.vecs [] def add(self, doc_id, text, vector): self.docs.append({id: doc_id, text: text}) self.vecs.append(vector) def search(self, query_vector, top_k3): rows [] for index, vec in enumerate(self.vecs): rows.append((cosine(query_vector, vec), self.docs[index])) rows.sort(keylambda x: x[0], reverseTrue) return rows[:top_k]当文档量到十万级以上时再替换成专门的开源向量库接口保持 add 和 search 两个方法不变上层不用动。这种先写最小实现、再平滑替换的路线能让你的故障面小很多。很多新人在第一步就把向量库配置、索引参数、权限、网络全搅在一起出了问题根本分不清是分块的问题还是数据库的问题。3.4 检索质量自测先看召回再看精度做完分块和 embedding 后别急着接大模型先用几个典型问题直接查向量库观察召回结果。我把最常见的坏现象整理成了一张表现象可能原因调整方向一条相关结果都没有分块粒度太粗/问题表述和文档用词差异大拆小块或对问题做改写后再 embedding召回结果看着相关但答不上来块里包含多个子话题语义向量被稀释缩小块长按段落语义边界分块每次召回的片段重复度过高重叠比例过大同一句出现在多个块降低重叠比例或者召回后做去重问题涉及多条件结果偏科单向量检索无法覆盖复合条件引入关键词过滤或混合召回必要时加 rerank这张表是对付“检索看起来能跑、实际效果稀烂”最有效的排查清单。我在联调阶段几乎每天都对着它看一遍先把这四类问题压下去再做完整问答链路。4. 把检索能力接到大模型完整的 RAG 问答链路实现4.1 用系统提示词让模型“按需决定是否查”现在进入正题让大模型自主决定要不要调用 MCP 工具。我的做法是在系统提示词里把工具描述和调用格式写清楚然后约定模型要么输出一个 JSON 动作要么直接给出最终回答。你是某公司内部知识库问答助手。当问题需要知识库中的事实支持时必须调用检索工具。 工具retrieve_knowledge 参数query为规范化的检索问句top_k为返回片段数默认3。 调用格式{tool: retrieve_knowledge, arguments: {query: 合同有效期, top_k: 3}} 如果不需要检索直接输出{reply: 你的回答}这里有个容易被忽略的细节query 参数不要直接用用户的原始提问而是要把它改写成适合检索的短句。用户可能说“我们那个合同一般签几年啊”更利于检索的 query 是“合同有效期”。在接入大模型之前可以先用规则或另一个模型做一次 query 改写RAG 的召回质量会明显上升。4.2 完整链路代码检索、注入、生成主链路的逻辑不复杂一共四步先让模型决定是否检索如果需要就把 query 发给 MCP 服务拿到检索片段后拼进上下文最后让模型基于参考资料给出答案。import json import urllib.request def chat_once(messages): # LLM_URL 指向你的模型服务网关此处假设返回格式兼容主流 chat 接口 req urllib.request.Request( os.environ[LLM_URL], datajson.dumps({messages: messages, temperature: 0.2}).encode(), headers{Content-Type: application/json}, ) resp json.loads(urllib.request.urlopen(req).read()) return resp[choices][0][message][content] def ask(question): messages [{role: system, content: SYSTEM_PROMPT}, {role: user, content: question}] first chat_once(messages) try: action json.loads(first) except json.JSONDecodeError: return first # 模型直接回答了 if tool in action and action[tool] retrieve_knowledge: tool_params action[arguments] context mcp_call(tools/call, {name: retrieve_knowledge, arguments: tool_params}) messages.append({role: assistant, content: first}) messages.append({role: user, content: build_context_prompt(context)}) return chat_once(messages) return action.get(reply, 抱歉我没能理解你的问题)注意这里我把第一次模型输出和检索结果都放进了消息历史再让模型继续回答。这样模型能看见自己刚才的决策过程回答时不会显得突兀。如果你不把工具调用结果放进去模型就会在没有任何检索依据的情况下强行回答RAG 和普通聊天就没有区别了。4.3 上下文模板让模型分清“引用”和“泛谈”检索结果拼进提示词时如果只是纯文本堆砌模型很容易分不清哪些是知识库内容、哪些是用户的话。我习惯用一个带边界的上下文模板。以下是知识库中检索到的参考资料用 doc 包裹的部分是原文只能作为回答依据不能原样照抄 doc 来源员工手册 内容事假需提前一天向直属主管提出申请请假流程以系统审批为准。 /doc doc 来源合同模板v3 内容标准服务合同的有效期为壹年自签署之日起计算。 /doc 请基于上述资料回答用户问题。如果资料不足以回答请直接说明“资料中未找到相关说明”。为什么要加来源字段一方面是为了回答时能带上引用另一方面是给后续排查召回问题留下追踪线索。我经常在回答质量下降时直接看召回片段来自哪些文档几秒就能定位是分块问题还是文档本身缺内容。4.4 一个能跑通的最小验证我用一个大约 300 条文档的内部手册测试集跑了一遍完整链路效果可以做到简单事实型问题比如“事假提前几天申请”模型会先调用检索工具把员工手册里的相关片段捞出来再给出带来源的答案。这个过程从问题输入到最终回答大概消耗两次模型调用延迟在 2 到 3 秒左右不算快但作为原型已经足够说明链路是通的。5. 联调阶段最容易出问题的三个位置5.1 MCP 服务的线程模型与超时处理联调时遇到的第一类问题就是请求卡住或者偶发连接被重置。原因往往是服务端用了单线程 HTTP Server而客户端在同一个连接里并发发起了多个请求。这个问题在本地测试时不容易暴露因为请求少、速度慢线程池压力小一旦多个 Agent 同时调用单线程服务端就会变成排队系统MCP 协议里的 JSON-RPC 消息 id 如果还重复异常会非常隐蔽。我的经验是服务端必须用线程化处理同时客户端要设置一个总超时比如 10 秒。超时后不要整个链路失败而是记录日志并把这次调用标记为失败让上层决定是重试还是放弃检索直接回答。RAG 最大的敌人不是答错而是迟迟没有响应。5.2 工具风暴大模型“疯狂调工具”怎么刹车第二个坑是模型陷入反复调用工具的循环。有一次我测试“简单寒暄”问题模型居然先调用检索工具查“今天天气怎么样”查到一篇无关的手册片段后又继续调用检索想找更多信息直到触发我的最大次数限制才停。这是因为我们把“遇到不懂就查”写进了系统提示词却没有给模型设置一个调用预算。正确的做法是添加两道保险第一限制单轮问答最多调用 N 次工具超过次数就强制进入生成阶段第二在提示词里明确只有涉及具体事实时才允许调用纯闲聊、推理、数学问题都不应该触发检索。把这两条加进去后工具风暴基本就不会再出现。5.3 上下文污染相关资料不是越多越好第三个坑最隐蔽召回片段过多导致的上下文污染。最初我把 top_k 调到 10以为资料给得越足模型回答越准结果恰恰相反。模型被迫在 10 段资料里“大海捞针”最终回答常常是把最显眼的那段话润色重写完全没有消化其他片段。更严重的是如果这 10 段里混入几段相似但不相关的文本模型有一定概率会采纳错误信息产生幻觉式回答。我后来把 top_k 压到 3 到 5并且对召回片段做了一遍去重和相关性过滤只保留和问题最直接相关的内容这才看到明显的质量提升。记住一个反直觉的结论RAG 的性能不是资料越多越好而是“刚好够用且无干扰”最好。6. 从 Demo 走向工程化性能、成本与多知识库隔离6.1 缓存与延迟预算原型能跑通以后你很快会开始关心两件事延迟和成本。我建议先建立一个简单的延迟预算表再决定哪些环节值得优化。环节典型耗时优化手段用户问题 embedding约 100ms对相同或相似 query 做缓存按天失效向量检索5ms 到 50ms换专用向量库、按索引分片首次模型调用约 1 秒无取决于模型服务检索后第二次模型调用约 1 秒可考虑把两次调用合并为一次但牺牲一定的决策灵活性embedding 缓存很便宜几乎每问必用同一批高频问题命中后能省掉一大截延迟。第二次模型调用也不是非做不可如果检索结果是强相关的你可以把检索条件直接写进用户消息里让模型在一个回合内完成回答省一次调用成本。6.2 多知识库隔离的 MCP 设计当知识库从 1 个变成 3 个、5 个MCP 工具的命名和隔离策略就变得重要。我见过有人把所有知识库合成一个工具只靠参数区分结果一个工具的参数列表越来越庞杂维护成本持续上升。更好的做法是用命名空间一个知识库一个工具比如 employee_handbook_search、contract_template_search每个工具的说明里写清楚适用场景。这样做的好处有三点。第一模型可以根据问题的语义精确选择知识库不会去错误的知识域里捞资料第二权限控制可以在服务端按工具粒度做不用在参数里传敏感信息第三每个工具的检索质量可以单独评测和迭代互不干扰。唯一需要注意的是别把工具名取得太长导致模型生成时被截断控制在合理长度内即可。6.3 评测集与验收线不能只靠感觉最后一步也是很多人会跳过的一步给 RAG 系统建一个评测集。没有评测集你调整分块参数、更换 embedding 模型时根本无法判断是变好了还是变坏了。我的做法是挑 50 到 100 条有代表性的问题每条问题记录它应该命中的文档和标准答案。评测时看两个指标召回命中率即检索结果里是否包含应该命中的片段回答准确率靠人工或者另一个模型判断最终回答是否在事实上和标准答案一致。我给自己定的验收线是召回命中率不低于 0.7回答准确率不低于 0.8低于这个线就回去调分块和检索策略不要急着上线。最后补一句我踩过的最深的坑不要在第一个版本里追求框架的完整性。很多人一开始就引入了大量重依赖结果整个调试过程都在和依赖打架。先手写 MCP 协议和向量检索的最小闭环把链路跑通、评测跑出基线再逐步换掉那些“只适合原型”的部件这条路走起来反而最快。希望这篇记录能帮你绕过前面那些坑少花几个晚上的无效调试时间。