这是MCP项目笔记的第十二篇专讲RAG-MCP。说起来这个项目从第一批基础工具开始一路做到现在最难啃的并不是协议对接本身而是怎么让模型在回答问题时“有据可依”。RAG-MCP简单理解就是按照MCP标准协议把检索增强生成RAG能力封装成一个独立服务让上层应用通过统一接口调用知识库检索再把检索结果拼进模型上下文。它解决的是模型知识时效性不足、私有知识无法参与推理的问题。这篇笔记适合正在做MCP服务端开发、想在工程里接入私有知识库、或者被“模型总在瞎编”折磨过的同学我尽量把实操链路和踩坑细节都写清楚。1. 为什么要做 RAG-MCP需求与价值拆解1.1 项目推进中遇到的上下文短板做这个项目之前我们的应用接入过不同的模型接口效果虽然能用但有几个问题始终绕不开一是模型的静态知识停留在训练时刻公司内部的制度文档、产品手册、历史故障记录它完全不知道二是就算我们尝试把文档塞进系统提示词几千token下去对话一长就超限而且关键信息容易被淹没三是每换一个模型服务商知识接入的代码就要重写一遍维护成本直线上升。后来我们意识到问题的本质不是“模型不够聪明”而是“模型缺少检索能力”。我们需要的不是一个更大的上下文窗口而是一条按需动态获取知识的通道。这条通道必须稳定、可复用、不依赖具体模型品牌。RAGRetrieval-Augmented Generation恰好提供了检索生成的框架但它本身不是一种连接标准工程上很难做到“一次接入处处可用”。这也是我们转向MCP的原因MCP定了客户端、服务端之间的通信规范把工具、资源、提示词都统一成标准能力RAG作为其中一个工具接入既解决了知识来源问题又顺手解决了多端复用问题。1.2 RAG 与 MCP 结合的核心收益把RAG封装成MCP服务之后最直观的收益是“接口标准化”。我们内部有两套对话应用、一套自动化报表系统需要用到知识问答以前每一套都要单独写一个检索函数现在只要在MCP客户端里注册同一个服务三套系统全部复用。第二个收益是“知识可治理”。检索索引、分块策略、重排序逻辑都收敛到服务端业务侧只关心传查询词、拿结果知识库更新时不需要通知各业务方服务端重启加载新索引即可。第三个收益可能不那么明显但实测下来很重要隔离了“模型调用”与“知识注入”的耦合度。我们在MCP服务里做了一次检索效果优化直接把重排序算法换掉客户端代码一行没改。如果把这个检索逻辑写死在应用层那这次改动至少要动三个项目。所以我的结论是当项目里已经存在多个应用共享知识库需求的时候确实值得把RAG单独做一层MCP服务而不是在应用里各自实现。2. RAG-MCP 的架构设计与组件拆解2.1 整体数据流从查询到知识注入先画一条完整链路方便后面看代码时知道自己站在哪个环节。用户在对话界面输入问题宿主应用先把问题发给大模型大模型在工具调用阶段决定是否调用search_knowledge工具这时候宿主应用通过MCP协议把查询词发送给RAG-MCP服务服务内部计算查询向量、在向量库里做相似度检索、把命中的文本片段按格式组装再返回给宿主宿主把这些片段注入上下文之后模型基于片段重新生成回答。简单说就是query进chunks出模型最后在chunks的基础上说人话。这里容易忽略的是“检索发生在模型调用的中途”而不是发生在用户输入之后。MCP的工具调用机制允许模型自主决定要不要检索、检索几次。这就比“每次对话都先检索一遍”要节省很多开销尤其在多轮对话里模型只在需要答案时才会触发工具。我们在设计时也刻意保留了这一点没有做成“强制每次必召回”效果明显更好。2.2 核心模块索引、检索、上下文组装RAG-MCP服务并不复杂核心模块可以拆成三块。索引模块负责把知识文档做分块、向量化和存储。分块是最容易影响检索质量的环节块太小语义不完整块太大噪声太多。我用的是先按标题结构切出一级段落再按固定窗口二次切分窗口大小512个字符、相邻块重叠48个字符。这个参数不是拍脑袋定的我拿线上50个高频问题测过召回率51248的组合比直接在段落级切分高出大约13个百分点原因很简单多数规章制度类问题在一个段落里就能覆盖答案但段落太长时查询向量容易被整体语义带偏小窗口配合少量重叠能稳住局部信息。检索模块负责把查询向量和候选向量做相似度计算取top结果。这块真正的难点不是算相似度而是让业务语义对齐。我们早期直接用原始向量内积效果很差后来把embedding输出做了L2归一化再用余弦相似度稳定很多。如果算力允许建议再多加一层轻量重排序拿top 20候选重新算一次得分最终返回top 5。上下文组装模块决定“模型最终看到什么”。不能把原始文档直接原样丢给模型要把命中片段去重、按相关度排序并标注来源编号。我们的输出格式是[引用#1 | 得分0.79] 根据内部规定第12条项目周期超过一个季度的需求必须拆分里程碑…… [引用#2 | 得分0.65] 变更流程需在系统内提交申请并抄送项目负责人……模型看到这种格式之后回答时会更倾向于引用具体条款而不是凭空编造。组装时还要控制总token数我们限定最多输出5个片段、单片段不超过350字基本能保证注入量在2000token以内。2.3 为什么不能把 RAG 写死进模型而是做成 MCP 服务有人可能觉得既然只是一个检索调用直接在应用里写个函数不行吗非要套一层MCP协议不是多此一举吗这问题我们内部也争论过。但在我们的真实场景里直接写死有三个硬伤第一各种业务方用的编程语言不一样有Python、Java、Node如果每个端都实现一份检索逻辑任何一个检索策略调整都要同步改N份代码第二知识库的权限控制、更新策略、访问审计全部无法统一不同应用接入的索引可能还不一致第三以后想换底层的向量存储方案影响面无限扩大。MCP把检索抽象成标准工具调用客户端只需要知道“工具名、参数、返回格式”这三件事不需要关心向量库部署在哪、用什么算法搜索。对我们这种“一套知识库服务多个业务”的团队来说这种解耦价值极高。注意如果你的场景只有一个小脚本、一个应用做完整的RAG-MCP可能确实有点重用普通函数就够了。判断标准很简单一旦出现“第二个消费方”就值得上MCP。3. 实操落地从零搭建一个 RAG-MCP 检索服务3.1 技术选型与仓库结构这里基于Python生态做一个最小可运行的RAG-MCP服务。选型原则是尽量少依赖方便后面替换。仓库结构是这样的rag-mcp/ ├── server.py # MCP服务入口定义工具 ├── index_service.py # 文档分块、向量化、入库 ├── vector_store.py # 简易向量存储内存版生产可替换 ├── embedding_client.py # 封装embedding模型调用 ├── docs_source/ # 待索引的原始文档 └── tests/ └── test_retrieval.py依赖只需要三样mcp官方SDK、numpy向量运算、requests调用embedding服务。需要注意MCP对Python版本有要求我用的3.10以上太老的版本没法装SDK。如果你不想装完整SDK也可以直接手写JSON-RPC通信但很容易在鉴权、会话管理上踩坑不建议。embedding模型我们没有单独本地化部署而是复用团队已有的模型服务接口对外只暴露一个HTTP接口传入文本返回归一化向量。这样能快速验证整体流程后续再换成本地模型只需要改embedding_client.py一个文件。3.2 实现向量化与索引入库先写embedding客户端。假设模型服务的请求和响应格式是固定的JSON# embedding_client.py import requests import numpy as np EMBEDDING_URL http://internal-embedding-service/v1/embed TIMEOUT_SECS 5 def get_embedding(text: str) - np.ndarray: resp requests.post( EMBEDDING_URL, json{text: text}, timeoutTIMEOUT_SECS, ) resp.raise_for_status() vector np.array(resp.json()[embedding], dtypenp.float32) # 归一化让内积结果等效于余弦相似度 norm np.linalg.norm(vector) if norm 0: vector vector / norm return vector然后实现一个简单的向量存储。生产环境我会建议用专业向量数据库但为了演示核心逻辑先用内存存储加线性扫描。这个实现对于几万条分块完全够用单次扫描在毫秒级别。代码如下# vector_store.py import numpy as np class VectorStore: def __init__(self): self.chunks [] self.vectors [] def add(self, chunk: str, vector: np.ndarray): self.chunks.append(chunk) self.vectors.append(vector) def search(self, query_vector: np.ndarray, top_k: int 5): if not self.vectors: return [] scores [float(np.dot(query_vector, vec)) for vec in self.vectors] top_indices np.argsort(scores)[-top_k:][::-1] return [(self.chunks[i], scores[i]) for i in top_indices]这里把查询向量也做了归一化因此内积就是余弦相似度。要注意如果你的embedding模型返回向量本身不是归一化的一定在存储和检索两端都做归一化否则召回排序会乱掉我测过不归一化时top 5结果里有3条是明显的偏题内容。再写索引服务负责把文档切块并入库# index_service.py from vector_store import VectorStore from embedding_client import get_embedding def split_text(text: str, chunk_size512, overlap48): if len(text) chunk_size: return [text] chunks [] start 0 while start len(text): end start chunk_size chunks.append(text[start:end]) if end len(text): break start end - overlap return chunks def build_index(store: VectorStore, docs: list[str]): for doc_id, text in enumerate(docs): chunks split_text(text) for idx, chunk in enumerate(chunks): vec get_embedding(chunk) store.add({doc_id: doc_id, chunk_id: idx, text: chunk}, vec)这里每次调用embedding服务都是同步的文档量大时建议加并发或异步不然索引速度感人。我们线上大概500份文档、切出8600多块单线程跑完花了将近20分钟后来改成一次批量提交128条文本时间压缩到3分钟以内。3.3 实现 MCP Tool检索与引用接下来是核心的MCP服务端。使用FastMCP的方式定义一个名为search_knowledge的工具宿主模型通过这个工具名发起检索。代码长这样# server.py from mcp.server.fastmcp import FastMCP from vector_store import VectorStore from embedding_client import get_embedding store VectorStore() # 假设索引在服务启动前已经加载完毕 # load_index(store) mcp FastMCP(RAG-MCP) mcp.tool() def search_knowledge(query: str, top_k: int 5) - str: 检索知识库返回与查询最相关的文档片段。 Args: query: 用户要查询的自然语言问题 top_k: 返回的片段数量默认5 query_vector get_embedding(query) results store.search(query_vector, top_k) lines [] for idx, (chunk, score) in enumerate(results, start1): lines.append(f[引用#{idx} | 得分{score:.2f}] {chunk}) return \n.join(lines) if __name__ __main__: # 使用stdio传输适配MCP客户端标准连接 mcp.run()这里有几个细节容易踩坑。第一工具函数的参数尽量写明确的Python类型和文档字符串因为MCP的SDK会把函数签名自动转成JSON Schema模型会根据字段描述判断怎么调用。如果你文档字符串写得太含糊模型传参经常不准确。第二mcp.run()默认走stdio传输适合本地子进程启动如果做远程调用建议换成HTTP/SSE模式但需要额外处理鉴权与网络策略。第三工具返回值目前是纯字符串好处是模型可以直接阅读坏处是丢失结构化字段。如果你希望后续自己拼结构化上下文可以返回JSON字符串看业务需要。3.4 客户端调用与上下文注入示例MCP客户端侧的调用逻辑因框架而异但核心路径类似客户端建立MCP连接列出工具列表在模型请求过程中允许工具调用。我这里写一个简化流程# client_demo.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def query_rag(question: str): server_params StdioServerParameters( commandpython, args[server.py], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() tool_names [t.name for t in tools] if search_knowledge not in tool_names: raise RuntimeError(工具未注册) result await session.call_tool( search_knowledge, arguments{query: question, top_k: 5}, ) return result.content实际应用里宿主会把result.content里的文本片段拼到系统提示词或者后续消息中再发给模型。我们自己的做法是加一个简单的系统注入模板你是一名内部知识助手。请优先依据以下参考资料回答问题。如果资料中没有相关内容明确回答“当前知识库中未找到相关信息”不要凭空编造。 参考资料 {检索到的片段}这个模板比单纯把片段丢给模型要稳很多尤其是“不编造”这一条极大减少了幻觉回答。我也建议在应用侧对工具执行前后的模型输出做个对比日志方便回溯是哪一次检索影响了最终答案。4. 常见问题与排查技巧实录4.1 高频问题速查表做这个项目时我们前后排过不少问题我把最典型、最容易反复踩的几个整理成表格。问题现象可能原因解决方法检索结果几乎不相关查询向量和文档向量不在同一语义空间确认查询端和索引端用的是同一个embedding模型且输入预处理一致相同问题的结果时好时坏索引更新后旧向量与新向量混合每次全量重建索引避免增量模型导致向量分布漂移模型不触发检索工具工具描述不清或返回格式不直观优化工具函数文档字符串返回结果里明确标注来源和得分上下文注入后模型答非所问注入片段过多、信噪比低降低top_k加重排序或者在提示词里强调“只参考高得分片段”调用工具时超时embedding服务耗时过长或网络波动在embedding客户端加超时重试并考虑批量embedding启动时端口冲突stdio模式被误配置成网络模式检查mcp.run()参数首选stdio避免端口管理最容易被忽视的是“同一模型、不同版本”的向量空间不一致。我们有一次手动更新了embedding模型的参数没有同步重建索引结果旧索引里全部是旧向量新查询用新向量去检索相似度全面偏低。之后我们规定凡是embedding模型有变更必须执行全量索引重建不允许混合新旧向量。4.2 检索质量优化分块、重排序与阈值检索质量是RAG-MCP的生命线。第一个优化方向是分块前面提到的51248是把“语义完整”和“定位精准”做了折中如果文档本身有明确小标题比如规章制度第几条强烈建议先用标题切成独立段落再对小段落做长度归一化。第二个方向是重排序先用向量召回top 20再用轻量级交叉编码器重新打分取top 5。交叉编码器对查询和文档的交互建模更充分效果提升非常明显但也更耗时所以只对候选集做。第三个方向是得分阈值。我们统计过真正有效的命中片段得分通常大于0.55如果top 5里的得分都低于该值那次检索大概率是无效的不如直接告诉模型“知识库中没有相关内容”减少误导。4.3 性能调优缓存、并发与批量在性能调优上我有三个实测经验。第一给embedding查询加缓存。同一问题在测试阶段或热门场景里会被反复问用LRU缓存做过结果缓存命中率大约25%平均响应时间从800毫秒降到300毫秒。第二embedding请求一定要并发。最初同步一条条请求10个候选就慢得让人崩溃改成ThreadPoolExecutor(max_workers8)批量请求之后top 20重排序的耗时缩短了70%。第三向量存储不要一股脑全放内存。虽然线性扫描在数据量小时很快但数据上到百万级就扛不住了建议引入专业的向量数据库并且把索引放到SSD上避免网络IO降低吞吐。另外MCP服务本身是无状态的可以轻松水平扩容。但这里有个细节如果每个实例都各自加载一份索引内存占用会线性增长。我们后来把索引服务独立出去向量数据库单独部署MCP服务只做代理和上下文组装这样每个实例都很轻扩容成本低很多。5. 下一步还能怎么扩展做完基础的RAG-MCP之后有一些明显的扩展方向。一个是可以给检索工具增加“指定文档范围”的参数比如doc_ids或category让模型能精确筛选知识源这对于有多个知识库权限隔离的场景非常有用。另一个是在返回片段时带上文档元信息比如更新时间、作者、标题让模型回答时能够给出更具体的来源说明用户追问时也能追溯到原文。如果你有对话记忆需求还可以在MCP服务端增加一个“最近检索历史”工具让模型在多轮对话里查看自己之前用过的检索词防止重复检索。我个人在实际操作中的体会是RAG-MCP本质上不是黑魔法它的架构价值大于算法价值。千万不要一上来就调模型先把分块、检索、注入这一段链路做通、做稳后面才谈得上效果优化。最后再分享一个小技巧调试RAG-MCP时不要只盯着工具返回内容看把MCP客户端发送的原始JSON-RPC请求也打出来有时候问题根本不在检索逻辑而是模型把参数传错了看到原始协议消息能省一多半排查时间。