RAG与MCP组合实现上下文感知智能代理的Python实践

今年做 AI 应用绕不开两个关键词RAG 和 MCP。RAG 负责把模型够不到的知识搬进对话上下文MCP 负责让模型能够按标准协议调用外部工具。当两者组合起来就是我们常说的上下文感知智能代理它既能从私有知识库里召回材料又能把召回结果作为工具返回值交给模型继续推理。这篇文章我会用 Python 从零实现这套流程代码不依赖重型框架核心路径你能完整看到。这套东西适合谁适合有 Python 基础、正在做 Agent、或者想给现有问答机器人加知识库的开发者。读完你能搞清楚知识库模块怎么切、MCP 服务怎么暴露工具、Agent 怎么决定调用工具、上下文怎么组织才不容易出幻觉。整个项目代码量不大但是把 RAG 和 MCP 的边界切得非常清楚后面换模型、换向量库、加新工具都不会伤筋动骨。1. 整体设计与思路拆解1.1 RAG 是什么MCP 是什么RAG全称 Retrieval-Augmented Generation检索增强生成。核心思路很直白模型不懂私有知识没关系先把相关资料查出来拼进提示词再让模型基于这些资料生成回答。这样不需要重新训练模型也不需要把全量知识塞进模型参数里。传统做法里你要么把所有知识写进提示词结果上下文很快就塞爆要么做微调结果知识更新一次要重新训练一次成本极高。RAG 绕开了这两个问题。每来一个问题现查现用知识更新只要更新文档库就可以了。MCP全称 Model Context Protocol模型上下文协议。它解决的是工具调用标准化的问题。在没有 MCP 之前Agent 要接一个工具就得写一套对接逻辑每个工具的请求格式、返回格式、错误处理都不一样。MCP 把一个工具封装成方法 入参描述 返回结果模型可以读取工具列表决定调用哪个工具然后通过统一格式拿到结果。我用一句话总结RAG 是知识层的管道MCP 是工具层的总线。两者解决的问题完全不同但组合起来正好补足了 Agent 的两块短板。1.2 为什么把 RAG 和 MCP 组合在一起先说痛点。你做一个内部问答机器人用户问报销流程是什么如果模型没有相关知识它大概率会一本正经地编一个流程这是典型的幻觉问题。RAG 能解决这个但光有 RAG 还不够因为问答场景往往还要查人、查系统、查天气、查工单这些数据不在静态文档里必须通过 API 去拿。如果每次接一个新数据源都写一套定制代码Agent 很快就变成一个维护成本极高的胶水工程。MCP 的作用就是把检索知识和调用系统统一成工具Agent 只需要学会一句话需要知识就调检索工具需要系统数据就调对应系统工具。更关键的是MCP 的工具描述是给模型看的。模型会基于工具名、描述、入参 schema 来判断要不要调用、怎么调用。检索工具一旦接入 MCP知识获取就变成 Agent 自身的一种能力而不是外部写好的一根固定管道。这个组合带来的直接效果是同一个 Agent面对不同问题可以自主决定这题直接回答还是先查资料再回答。这就是上下文感知智能代理的雏形。1.3 技术选型我为什么这么选这个项目里我刻意没有上重框架核心依赖只有几样FastAPI 做 MCP 服务的 HTTP 传输层httpx 做异步请求numpy 做相似度计算jieba 做中文分词。向量库也没有用独立服务而是用 Python 类自己维护一个内存版方便你看懂检索过程。这样的选型有几个好处。第一能复现。你不需要装一个庞大的向量数据库也不需要申请额外的云服务代码拉下来就能跑。第二边界清楚。一旦出了问题你能直接判断是检索模块的问题还是 Agent 编排的问题。第三替换成本低。我在代码里把向量化做成了一个函数生产环境只要把函数内部换成真正的 embedding 接口其他部分几乎不用改。我见过不少项目一开始就上全套框架结果 Agent 不干活时根本不知道是模型不行还是框架配置错了。所以我强烈建议第一版把手写透后面再迁框架你会顺手得多。2. 核心细节解析与实操要点2.1 上下文感知智能代理的模块划分整个系统我拆成五个模块每个模块只做一件事模块职责关键点知识加载与切片把文档变成固定大小的文本块保留标题和来源信息向量化与检索把文本块和用户查询变成向量算相似度用余弦相似度取 top_kMCP 工具服务把检索能力包装成统一工具提供工具描述和入参 schemaAgent 核心让模型决定要不要调工具并组织最终回答维护系统提示词、历史、工具结果对话记忆保存最近若干轮对话只保留近 N 轮避免上下文膨胀模块切得细不是说代码要多复杂而是为了让每一段逻辑都拥有独立的测试边界。比如知识库检索你不接 Agent 也能单独测试MCP 服务你不接知识库也能单独测试。等到联调出问题时你只需要逐段定位。2.2 向量化与检索到底怎么做很多新手以为 RAG 就是把文档存进数据库然后搜索。实际上真正决定检索质量的是两件事文本切片和向量相似度。文本切片不能盲目按固定字数硬切。如果一刀切把一个完整段落切开模型拿到的是半截内容回答必然不准。我建议按段落结构切同时保留一定重叠。参数建议值说明切片大小300500 字太小则语义不完整太大则噪声多重叠长度4080 字保证跨段上下文不断裂来源信息必须保留方便 Agent 回答时引用出处向量化部分生产环境基本都用 embedding 模型把文本映射成一个固定长度的向量。两个文本语义越接近向量夹角越小余弦相似度越高。我这份代码里为了不引入重型依赖用 jieba 分词加 TF-IDF 做了一个简易版向量效果够演示但不代表生产水平。你把这个函数替换成真正的 embedding 模型调用整个检索链路立刻会质变。检索的时候有个细节不要只返回相似度最高的第一段。很多时候答案分散在多个段落里或者第一段落只是开头、真正要点在后面。我会返回 top_k默认 3 段让模型综合判断。2.3 MCP 服务到底暴露什么能力MCP 的核心是三个概念工具列表、工具调用、统一结果返回。模型启动后会拿到一份工具列表里面包含工具名、工具描述、入参结构。当模型认为当前问题需要调用工具时它会返回一个工具调用指令Agent 再发起实际的工具调用。我这里实现了一个极简版 HTTP MCP 服务提供了两个方法tools/list返回工具列表tools/call执行具体工具。工具列表里每个工具都有name和description这两个字段非常关键因为模型是靠它们来判断什么时候调用的。举个例子retrieve_knowledge的描述我写的是从本地知识库检索与 query 相关的段落用于回答需要内部资料的问题。描述写得太泛模型会把它当成搜索引擎什么都调写得太窄模型该调的时候不调。这个描述要反复调试。返回结果我统一用 MCP 的 content 数组格式文本内容放在content[0].text里。Agent 拿到这个返回值后会把它作为一条tool消息回填给模型模型再基于工具结果生成最终回答。2.4 上下文组装别把模型当无限容量用上下文感知不等于把所有东西全塞给模型。系统提示词、历史对话、检索结果、用户问题这四部分都有自己的位置。我的组装顺序是先放系统提示词再放最近 8 轮历史对话然后是当前用户问题。当 Agent 调用完工具再把工具结果追加上去让模型在第二次推理时看到完整链路。这里有几个容易踩的坑。第一历史对话不是越多越好超过容量上限后最早的历史会被截断但检索结果反而不能随便丢因为那是当前问题的关键材料。第二检索结果要控制数量我默认 top_k3超过这个量模型容易捡了芝麻丢了西瓜。第三系统提示词里要明确告诉模型有来源标记的内容可以直接引用没有来源的内容不要编。我习惯在代码里留一个MAX_CONTEXT_CHARS之类的参数对历史对话做保护性截断而不是粗暴地截断检索结果。具体逻辑后面代码里会体现。3. 实操过程与核心环节实现3.1 安装依赖和环境准备这个项目依赖很少Python 3.9 以上就行。先装依赖pip install fastapi uvicorn httpx numpy jieba然后准备一个知识目录。我建议叫knowledge/里面放一些纯文本文件比如qa.txt。内容随便写一些你希望 Agent 掌握的私有知识比如制度文档、产品说明、操作手册。代码启动时会自动读取这个目录下的所有 txt 文件。如果你用的是大模型服务商的兼容接口需要设置环境变量LLM_BASE_URL指向 chat completions 接口地址LLM_API_KEY填密钥LLM_MODEL填模型名。没有密钥也可以跑通除最终生成外的所有环节因为检索和 MCP 部分不依赖大模型。3.2 实现一个能做语义检索的知识库新建rag_kb.py。先实现文档切片import re def split_text(text, chunk_size500, overlap60): paragraphs re.split(r\n, text.strip()) chunks [] current for para in paragraphs: if not para.strip(): continue if len(current) len(para) chunk_size: current para \n else: if current: chunks.append(current.strip()) current # 超长段落按窗口切 while len(para) chunk_size: chunks.append(para[:chunk_size]) para para[chunk_size - overlap:] current para \n if current.strip(): chunks.append(current.strip()) return chunks然后实现简易知识库类。这里用 jieba 做分词用 TF-IDF 做词频向量最终用余弦相似度检索import math from collections import Counter import jieba import re class SimpleKB: def __init__(self): self.chunks [] self.metadatas [] self.df_table Counter() self.total_docs 0 def _tokenize(self, text): return [ w for w in jieba.lcut(text.lower()) if w.strip() and not re.fullmatch(r[\W_], w) ] def _tfidf(self, tokens): counter Counter(tokens) vec {} for token, cnt in counter.items(): df self.df_table.get(token, 1) vec[token] cnt * math.log((self.total_docs 1) / (df 1)) 1 return vec def add_document(self, text, source): for chunk in split_text(text): tokens self._tokenize(chunk) self.total_docs 1 for token in set(tokens): self.df_table[token] 1 self.chunks.append(chunk) self.metadatas.append({source: source}) def search(self, query, top_k3): q_vec self._tfidf(self._tokenize(query)) scores [] for i, vec in enumerate(self._current_vecs()): common set(q_vec.keys()) set(vec.keys()) if not common: scores.append((i, 0.0)) continue dot sum(q_vec.get(t, 0) * vec.get(t, 0) for t in common) norm_q math.sqrt(sum(v * v for v in q_vec.values())) norm_d math.sqrt(sum(v * v for v in vec.values())) scores.append((i, dot / (norm_q * norm_d 1e-6))) scores.sort(keylambda x: x[1], reverseTrue) return [ (self.chunks[i], self.metadatas[i], s) for i, s in scores[:top_k] ] def _current_vecs(self): # 为了方便阅读这里在每次查询时重新计算向量 # 生产环境应该预先计算好并缓存 return [self._tfidf(self._tokenize(chunk)) for chunk in self.chunks]这段代码里最关键的是search函数。它把查询词向量和每个切片的向量做余弦相似度计算返回 top_k 结果。需要说明的是_current_vecs每次查询都重算效率很低只适合教学演示。生产环境中应该在建库时就完成向量化并落库。真正的 embedding 模型接入点也在这里。你只需要把_tfidf换成调用 embedding 接口返回一个固定长度的数值列表然后把余弦相似度改成 numpy 点积其他逻辑不变。3.3 实现一个极简 MCP 服务新建mcp_server.py。这个文件负责加载知识库并把检索能力暴露成 MCP 工具import os import json from fastapi import FastAPI, Request from rag_kb import SimpleKB, split_text app FastAPI() kb SimpleKB() # 启动时加载 knowledge 目录下的所有文本 KNOWLEDGE_DIR ./knowledge def load_knowledge(): if not os.path.exists(KNOWLEDGE_DIR): os.makedirs(KNOWLEDGE_DIR) for filename in os.listdir(KNOWLEDGE_DIR): if filename.endswith(.txt): path os.path.join(KNOWLEDGE_DIR, filename) text open(path, encodingutf-8).read() kb.add_document(text, sourcefilename) load_knowledge() def make_result(text, req_id1): return { jsonrpc: 2.0, id: req_id, result: { content: [ {type: text, text: text} ] } } app.post(/rpc) async def rpc(request: Request): payload await request.json() req_id payload.get(id, 1) method payload.get(method) if method tools/list: return { jsonrpc: 2.0, id: req_id, result: { tools: [ { name: retrieve_knowledge, description: 从本地知识库检索与 query 相关的段落用于回答需要内部资料的问题。, inputSchema: { type: object, properties: { query: {type: string, description: 用户的查询关键词或问题}, top_k: {type: integer, description: 返回段落数量默认 3} } } } ] } } if method tools/call: params payload.get(params, {}) tool_name params.get(name) arguments params.get(arguments, {}) if tool_name retrieve_knowledge: query arguments.get(query, ) top_k int(arguments.get(top_k, 3)) results kb.search(query, top_ktop_k) if not results: return make_result(未检索到相关内容, req_id) parts [] for chunk, meta, score in results: source meta.get(source, 未知来源) parts.append(f[来源:{source}]\n{chunk}) return make_result(\n\n.join(parts), req_id) return { jsonrpc: 2.0, id: req_id, error: {code: -32601, message: f未知工具: {tool_name}} } return { jsonrpc: 2.0, id: req_id, error: {code: -32601, message: f未知方法: {method}} }启动方式uvicorn mcp_server:app --host 127.0.0.1 --port 8000你可以先用 curl 验证一下tools/list是否正常curl -X POST http://127.0.0.1:8000/rpc \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:1,method:tools/list}返回结果里应该能看到retrieve_knowledge这个工具。3.4 实现代理主流程从用户输入到工具调用新建agent.py。这里要做三件事维护系统提示词和历史对话、调用大模型的 function calling 能力、在模型需要时通过 MCP 服务获取检索结果。import os import json import asyncio import httpx MCP_URL http://127.0.0.1:8000/rpc LLM_BASE_URL os.getenv(LLM_BASE_URL, https://your-llm-endpoint/v1/chat/completions) LLM_API_KEY os.getenv(LLM_API_KEY, ) LLM_MODEL os.getenv(LLM_MODEL, your-model-name) SYSTEM_PROMPT 你是一个上下文感知的智能助手。你有以下工具可以调用 - retrieve_knowledge(query, top_k): 查询内部知识库返回相关文本段落。当问题涉及公司内部制度、产品文档、操作手册时必须先调用这个工具再根据查到的内容回答。 如果工具调用结果中有[来源:xxx]这样的标志说明这些内容是知识库检索到的你可以直接引用。回答要简洁、准确不要编造知识库里没有的内容。 def build_messages(history, user_input): messages [{role: system, content: SYSTEM_PROMPT}] # 只保留最近 8 轮防止上下文被历史对话撑爆 messages.extend(history[-8:]) messages.append({role: user, content: user_input}) return messages async def llm_chat(messages): payload { model: LLM_MODEL, messages: messages, temperature: 0.2, tools: [ { type: function, function: { name: retrieve_knowledge, description: 从本地知识库检索与 query 相关的段落。, parameters: { type: object, properties: { query: {type: string}, top_k: {type: integer, default: 3} }, required: [query] } } } ], tool_choice: auto } headers {Authorization: fBearer {LLM_API_KEY}} async with httpx.AsyncClient() as client: resp await client.post(LLM_BASE_URL, jsonpayload, headersheaders, timeout60) resp.raise_for_status() return resp.json() async def call_mcp_tool(tool_name, arguments): payload { jsonrpc: 2.0, id: 1, method: tools/call, params: {name: tool_name, arguments: arguments} } async with httpx.AsyncClient() as client: resp await client.post(MCP_URL, jsonpayload, timeout30) resp.raise_for_status() data resp.json() if error in data: raise RuntimeError(data[error].get(message, MCP 调用失败)) content data[result][content] return \n.join(item.get(text, ) for item in content if item.get(type) text) async def agent_run(history, user_input): messages build_messages(history, user_input) for _ in range(3): data await llm_chat(messages) msg data[choices][0][message] if msg.get(tool_calls): messages.append(msg) for tc in msg[tool_calls]: fn tc[function] args json.loads(fn.get(arguments) or {}) tool_output await call_mcp_tool(fn[name], args) messages.append({ role: tool, tool_call_id: tc[id], content: tool_output }) continue return msg[content] return 我尝试了几次工具调用后仍然没有得到可靠答案请稍后再试。 async def main(): history [] print(输入 exit 退出) while True: user_input input(你: ) if user_input.lower() exit: break reply await agent_run(history, user_input) print(AI:, reply) history.append({role: user, content: user_input}) history.append({role: assistant, content: reply}) if __name__ __main__: asyncio.run(main())这段代码里最核心的是agent_run里的循环。模型第一次返回可能不直接给答案而是带tool_calls字段。Agent 看到这个字段就会把模型要求的工具调用执行掉把结果作为tool角色消息追加到 messages再调一次模型。整个过程最多循环 3 次避免模型陷入无限调工具的死循环。3.5 把代码跑起来假设你在knowledge/qa.txt里写了一句话报销流程员工先填写报销申请单由部门负责人审批然后提交财务复核最后打款。启动顺序是# 终端 1启动 MCP 服务 uvicorn mcp_server:app --host 127.0.0.1 --port 8000 # 终端 2运行 Agent python agent.py然后输入报销流程是什么。模型会认为这个问题涉及内部制度于是调用retrieve_knowledge拿到知识库返回的段落再基于段落生成回答。整个过程你在代码里能看到两次 LLM 调用第一次是模型决定调工具第二次是模型基于工具结果回答。如果你的模型服务商不支持 function calling代码就需要降级在agent_run里省掉tools参数直接先调用retrieve_knowledge再把检索结果拼进 system prompt。效果会差一些但至少管道是通的。4. 常见问题与排查技巧实录4.1 依赖冲突和版本兼容FastAPI 和 pydantic 版本比较敏感。如果你之前装过老版本 pydantic启动mcp_server.py时可能会报字段校验异常。解决办法是先升级pip install --upgrade fastapi uvicorn pydantic另外 jieba 在 Python 3.12 以上可能会遇到找不到pkg_resources的报错这是因为 setuptools 没装。加一句pip install setuptools这类问题不算项目 bug基本就是环境问题。我的建议是尽量建一个新虚拟环境不要和别的项目混在一起。4.2 检索结果和问题对不上如果你发现 Agent 明明调了retrieve_knowledge回答还是胡说先别怀疑模型去检查检索结果本身。一个常见原因是切片太粗一个 chunk 里揉进了多个知识点。另一个原因是 top_k 返回太少真正有效的段落排在第 4、第 5 名。调试方法很简单直接调用 MCP 服务不看 Agentcurl -X POST http://127.0.0.1:8000/rpc \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:1,method:tools/call,params:{name:retrieve_knowledge,arguments:{query:报销流程,top_k:5}}}看返回内容是否包含关键答案。如果返回的内容和问题对不上核心问题在检索不在 Agent。生产环境里我一般会先换更强的 embedding 模型再不行就上 rerank 模型最后才考虑调切片策略。4.3 上下文超限上下文超限是最容易遇到的报错尤其当你把十几轮历史对话和检索结果全塞进 messages 时。模型对 token 数有硬上限超了就直接报错。我的策略是历史对话只保留最近 8 轮且每轮回答长度也做限制。检索结果最多给 3 段超过 3 段的信息往往不是帮助模型而是干扰模型。还有一个细节中文和英文切出来的 token 数不一样不要用字符长度去估算 token 数最好在代码里加一个 tokenizer 或按 1.5 倍字符数做保守估算。4.4 MCP 调用超时或没有返回MCP 调用超时原因往往不是代码逻辑而是服务根本没起来。先确认 8000 端口是否监听curl -X POST http://127.0.0.1:8000/rpc \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:1,method:tools/list}如果返回 404检查 uvicorn 是否真的启动成功启动命令是否带--port 8000。另一个常见问题是arguments被序列化成了字符串而不是 JSON 对象。在 Agent 代码里我专门用了json.loads解析 function calling 返回的 arguments这个坑几乎人人都会踩。4.5 常见问题速查表现象可能原因解决办法模型不调用工具工具描述写得模糊或模型不支持 function calling把描述改具体或降级为先检索后生成检索结果为空knowledge 目录没放文件或路径不对启动时检查load_knowledge是否真的读到了文件回答编造内容检索结果没被拼进第二次请求检查messages里是否有role: tool消息上下文超限历史对话或检索结果太长限制最近 8 轮top_k 设为 3加截断保护MCP 调用超时MCP 服务没启动或参数不对curl 单测tools/list和tools/call相似度全是 0切片内容与查询词没有公共词demo 版向量太弱换成真实 embedding 模型最后再说一个实际落地时的小习惯先把 MCP 服务和知识库拆开测再跑 Agent。因为 Agent 一旦接入工具同一个报错可能是模型问题、工具问题、上下文问题三者的叠加。我见过太多人花两小时调模型 prompt最后一查发现是知识库文件根本没有加载进去。写 Agent 和写传统后端不一样你要学会把链路切成一截一截验证哪一截断了先补哪一截这个习惯能帮你少走很多弯路。