企业官网建设流程全解析

先抛一个我最近反复跟人讲的观点：RAG 系统好不好用，七成取决于检索那一环做得怎么样。而我这次从零搭建基于 MCP 的 RAG 系统，最大的收获不是调通了一个大语言模型问答 Demo，而是把“检索”从业务代码里彻底拆出来，变成了一个可以通过标准协议访问的能力。本文会老老实实记录这个最小闭环的完整搭建过程，包括手写 MCP 服务、分块和 embedding 的细节、完整问答链路代码，以及联调时踩过的三个大坑。适合已经跑过简单 RAG、想进一步把检索服务化的开发者参考，也适合想搞懂 MCP 协议到底怎么用的新人。

1. 这个 Demo 想解决的痛点：RAG 里的检索为什么需要 MCP

1.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/call

MCP 的完整规范里有很多细节，但我们的 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(f"unknown 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, params=None): 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", data=payload, 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_k=3): rows = [] for index, vec in enumerate(self.vecs): rows.append((cosine(query_vector, vec), self.docs[index])) rows.sort(key=lambda x: x[0], reverse=True) return rows[:top_k]

当文档量到十万级以上时，再替换成专门的开源向量库，接口保持 add 和 search 两个方法不变，上层不用动。这种先写最小实现、再平滑替换的路线，能让你的故障面小很多。很多新人在第一步就把向量库配置、索引参数、权限、网络全搅在一起，出了问题根本分不清是分块的问题还是数据库的问题。

3.4 检索质量自测：先看召回再看精度

做完分块和 embedding 后，别急着接大模型，先用几个典型问题直接查向量库，观察召回结果。我把最常见的坏现象整理成了一张表：

这张表是对付“检索看起来能跑、实际效果稀烂”最有效的排查清单。我在联调阶段几乎每天都对着它看一遍，先把这四类问题压下去，再做完整问答链路。

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"], data=json.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 缓存很便宜，几乎每问必用同一批高频问题，命中后能省掉一大截延迟。第二次模型调用也不是非做不可，如果检索结果是强相关的，你可以把检索条件直接写进用户消息里，让模型在一个回合内完成回答，省一次调用成本。

6.2 多知识库隔离的 MCP 设计

当知识库从 1 个变成 3 个、5 个，MCP 工具的命名和隔离策略就变得重要。我见过有人把所有知识库合成一个工具，只靠参数区分，结果一个工具的参数列表越来越庞杂，维护成本持续上升。更好的做法是用命名空间：一个知识库一个工具，比如 employee_handbook_search、contract_template_search，每个工具的说明里写清楚适用场景。

这样做的好处有三点。第一，模型可以根据问题的语义精确选择知识库，不会去错误的知识域里捞资料；第二，权限控制可以在服务端按工具粒度做，不用在参数里传敏感信息；第三，每个工具的检索质量可以单独评测和迭代，互不干扰。唯一需要注意的，是别把工具名取得太长导致模型生成时被截断，控制在合理长度内即可。

6.3 评测集与验收线：不能只靠感觉

最后一步，也是很多人会跳过的一步：给 RAG 系统建一个评测集。没有评测集，你调整分块参数、更换 embedding 模型时，根本无法判断是变好了还是变坏了。

我的做法是挑 50 到 100 条有代表性的问题，每条问题记录它应该命中的文档和标准答案。评测时看两个指标：召回命中率，即检索结果里是否包含应该命中的片段；回答准确率，靠人工或者另一个模型判断最终回答是否在事实上和标准答案一致。我给自己定的验收线是：召回命中率不低于 0.7，回答准确率不低于 0.8，低于这个线就回去调分块和检索策略，不要急着上线。

最后补一句我踩过的最深的坑：不要在第一个版本里追求框架的完整性。很多人一开始就引入了大量重依赖，结果整个调试过程都在和依赖打架。先手写 MCP 协议和向量检索的最小闭环，把链路跑通、评测跑出基线，再逐步换掉那些“只适合原型”的部件，这条路走起来反而最快。希望这篇记录能帮你绕过前面那些坑，少花几个晚上的无效调试时间。