之前带过不少想入门 AI 大模型应用开发的朋友发现大家最常遇到的问题并不是“看不懂论文”而是不知道从哪里开始动手。网上资料要么是纯概念科普看完还是不会写代码要么是某个框架的碎片教程换个场景就抓瞎。这篇文章我打算把 AI 大模型应用开发最核心的三块内容——RAG、Agent、MCP串成一条完整的学习主线从零开始讲原理、写代码、跑通一个可演示的项目。无论你是刚接触 Python 的后端开发、想转 AI 应用方向的学生还是已经在业务里尝试接入大模型的工程师这篇文章都能帮你建立一套相对完整的知识框架。以下内容会按照“概念 → 环境 → RAG → Agent → MCP → 全栈实战 → 排错 → 最佳实践”的顺序展开。代码以 Python 为主会给出完整的可运行示例。所有项目会尽量保持模块化方便你根据自己的业务场景裁剪使用。1. 为什么要同时学 RAG、Agent 和 MCP很多人会把 RAG、Agent、MCP 混在一起聊但它们其实是三个不同层面的问题。只有把三者分开理解再组合起来才能真正明白大模型应用开发的完整路径。1.1 RAG解决模型“不知道”的问题大模型在训练完成之后知识就固定了。这里有两个致命问题知识滞后模型不会自动知道昨天新发布的产品文档。幻觉问题模型对自己不知道的内容可能一本正经地用不存在的知识回答你。RAGRetrieval-Augmented Generation检索增强生成的思路是提问之前先从外部知识库里检索出和问题相关的片段把片段拼进提示词再让大模型基于这些片段作答。相当于给模型开卷考试而不是闭卷瞎编。RAG 的核心流程可以简化成五步把文档切分成小块chunk。用嵌入模型Embedding Model把每一块转成向量。把向量和文本一起存进向量数据库。用户提问时把问题也转成向量做相似度检索。把检索结果和问题一起发给大模型生成最终答案。1.2 Agent解决模型“不能做”的问题普通聊天只能“说”Agent智能体的意义在于“做”。一个 Agent 通常包含以下能力规划Planning把复杂任务拆成多个步骤。工具调用Tool Use通过函数调用获取数据、操作系统、读写文件、查询数据库。记忆Memory在多次交互中记住上下文。反思Reflection执行失败后重新决定下一步行动。比较经典的 Agent 工作流是 ReAct模型先推理Reasoning出当前该做什么然后调用工具Acting拿到工具结果后再继续推理直到任务完成。1.3 MCP解决工具接入标准问题当你想让大模型调用工具时最原始的做法是“写死在代码里”def call_tool(name, args): if name get_weather: return weather_api(args[city])这种方式在只有两三个工具时很清晰但在复杂项目里会出现问题每个工具都要单独封装、协议不统一、权限模型混乱、调试困难。MCPModel Context Protocol模型上下文协议就是来解决这个问题的标准化协议。MCP 的架构可以类比成“AI 应用的 USB-C 接口”MCP Server提供工具、资源或提示词比如“数据库查询工具”“GitHub 操作工具”。MCP Client连接 Server 的一方通常运行在大模型应用侧。工具发现机制客户端可以动态获取 Server 支持的工具列表、参数结构不再硬编码。一个标准的 MCP Server 可以通过stdio或sse传输方式运行。2026 年的今天MCP 已经不只是实验性概念不少浏览器扩展、调试工具、数据库客户端都开始提供 MCP Server 接口。这也是为什么现在 AI 应用开发面试里MCP 几乎是必问题。1.4 三者的关系可以把三者的关系总结成一句话RAG 管知识Agent 管决策与行动MCP 管工具接入标准。一个完整的大模型应用通常是“RAG 提供知识基础 → Agent 负责拆解任务 → MCP 负责对接各种工具”。后面第 5 章的实战项目会把三者组合在一起演示。2. 环境准备与学习路线2.1 版本与工具说明由于大模型相关库迭代速度非常快我不建议给出具体的硬编码版本号。本文示例以以下环境为基准但你实际使用时需要根据项目情况调整组件推荐选择说明操作系统Windows 10/11、macOS、Linux 均可本文代码是跨平台的Windows 使用python -m venv创建虚拟环境Python3.10 以上推荐 3.11 或 3.12对类型注解和异步支持更好包管理pip 或 uv建议先掌握 pipuv 后续再学IDEVS Code / PyCharm推荐 VS Code Python 插件大模型 APIOpenAI 兼容接口国内厂商、本地模型Ollama/vLLM大多提供 OpenAI 兼容接口向量数据库Chroma本地开发/ FAISS / Milvus生产本文先用 Chroma零配置最容易跑通嵌入模型本地 sentence-transformers 或 API 嵌入演示用本地小模型避免额外费用2.2 依赖安装创建一个新的项目目录并初始化虚拟环境mkdir ai-learning-hub cd ai-learning-hub python -m venv venvWindows 激活虚拟环境venv\Scripts\activatemacOS / Linuxsource venv/bin/activate安装基础依赖pip install fastapi uvicorn openai chromadb langchain-core langchain-community python-dotenv httpx sse-starlette这些库的作用fastapi和uvicorn用于搭建 Web 服务。openai调用 OpenAI 兼容接口。chromadb本地向量数据库。langchain-core和langchain-community提供链式抽象和文档加载工具。httpx异步 HTTP 客户端后续 Agent 调用工具用。sse-starlette实现 SSE 流式输出。2.3 项目总结构整个实战项目我建议按下面结构组织ai-learning-hub/ ├── .env ├── requirements.txt ├── app/ │ ├── __init__.py │ ├── config.py │ ├── llm.py │ ├── embeddings.py │ ├── vector_store.py │ ├── rag.py │ ├── tools.py │ ├── agent.py │ ├── mcp_server.py │ ├── mcp_client.py │ └── main.py ├── data/ │ └── docs/ # 放你的知识库文档 └── tests/ └── test_rag.py后面的代码会按这个路径组织。为了便于理解每一小节会给出文件路径方便你对照建文件。3. RAG 实战从零搭建一个本地知识库问答系统这一节先不依赖 LangChain而是手动实现一个最小 RAG 流程。只有理解了每一步在做什么后面用框架时才知道哪里出了问题。3.1 文档切块Chunking为什么要切块因为大模型有上下文窗口限制而且把整篇几万字的文档塞进去效果不一定好。检索时我们希望找到与问题最相关的那一两段而不是整篇文章。切块的策略有很多最简单的是按固定字数切并且上下两块之间保留重叠区域。# 文件路径app/rag.py部分代码 from typing import List def split_text(text: str, chunk_size: int 500, overlap: int 50) - List[str]: 将长文本按 chunk_size 切块相邻块之间保留 overlap 字符重叠。 重叠可以避免关键信息恰好落在两个块的边界。 if len(text) chunk_size: return [text] chunks [] start 0 while start len(text): end min(start chunk_size, len(text)) # 尽量在段落或句子边界处切分 if end len(text): # 从 end 往前找最近的换行符 last_newline text.rfind(\n, start, end) if last_newline ! -1 and end - last_newline 100: end last_newline 1 chunks.append(text[start:end]) start max(end - overlap, 1) return chunks这里有一个关键点切块大小没有绝对标准。如果业务文档是 FAQ 类可以直接按条目切如果是长篇技术文档500 字左右通常比较合适。块太大检索出来容易超过上下文窗口块太小语义不完整检索准确率会下降。3.2 嵌入与向量化嵌入Embedding是把文本变成一组浮点数向量的过程。语义接近的两个句子它们在向量空间中的距离也接近。这里使用 sentence-transformers 加载本地小模型# 文件路径app/embeddings.py from sentence_transformers import SentenceTransformer _model None def get_embedding_model(): global _model if _model is None: # 这是一个轻量级中文嵌入模型首次运行会自动下载 _model SentenceTransformer(shibing624/text2vec-base-chinese) return _model def embed_text(text: str): model get_embedding_model() return model.encode(text).tolist() def embed_texts(texts): model get_embedding_model() return model.encode(texts).tolist()注意直接用pip install sentence-transformers安装即可。如果你的环境无法访问 Hugging Face比如网络策略限制可以把模型文件提前下载到本地然后通过SentenceTransformer(本地路径)加载。本地加载方式适合企业内部知识库场景。3.3 向量存储与检索Chroma 是一个非常好用的本地向量数据库。它会自动把向量和文档内容持久化在本地目录。# 文件路径app/vector_store.py import chromadb from chromadb.config import Settings class VectorStore: def __init__(self, persist_dir./chroma_db, collection_namedocs): self.client chromadb.PersistentClient( pathpersist_dir, settingsSettings(anonymized_telemetryFalse) ) self.collection self.client.get_or_create_collection( namecollection_name, metadata{hnsw:space: cosine} # 使用余弦距离 ) def add_documents(self, docs: list[str], metadatas: list[dict] | None None): ids [fdoc_{i}_{hash(doc)} for i, doc in enumerate(docs)] self.collection.add( idsids, documentsdocs, metadatasmetadatas or [{}] * len(docs) ) def search(self, query_embedding: list[float], top_k: int 3): result self.collection.query( query_embeddings[query_embedding], n_resultstop_k, ) return result[documents][0], result[metadatas][0]有几个容易踩坑的地方距离函数文本检索一般用余弦距离cosine不要选默认的 L2。ID 唯一性相同文档重复插入时 ID 不能冲突否则会覆盖或被拒绝。metadata 数量metadatas数量必须和documents一致。3.4 构建 prompt 并调用大模型检索到相关片段后下一步是构造 prompt。这里用一个简单的模板# 文件路径app/rag.py接前面 from openai import OpenAI def build_prompt(question: str, contexts: list[str]) - str: context_text \n\n.join([f[资料{i1}]\n{ctx} for i, ctx in enumerate(contexts)]) return f你是企业内部知识助手请严格依据参考资料回答问题。 如果参考资料中找不到答案请直接说明“资料库中暂无相关内容”不要编造。 参考资料 {context_text} 用户问题{question} 请用简洁、准确的中文回答 def ask_rag(question: str, vector_store: VectorStore, embedding_model, client: OpenAI, top_k: int 3): # 1. 把问题转成向量 query_vec embedding_model.encode(question).tolist() # 2. 检索 docs, metadatas vector_store.search(query_vec, top_ktop_k) # 3. 构造 prompt prompt build_prompt(question, docs) # 4. 调用大模型 response client.chat.completions.create( modelgpt-4o-mini, # 根据你的模型服务商调整 messages[{role: user, content: prompt}], temperature0.3, ) return response.choices[0].message.content, docs如果是本地部署的大模型只需要修改base_url和model。比如使用 Ollamaclient OpenAI( base_urlhttp://localhost:11434/v1, api_keyollama, # Ollama 本地不需要真实 key )把model换成你本地拉取的模型名比如qwen2.5:7b代码其他部分完全不用改。这就是 OpenAI 兼容接口的价值。3.5 本地知识库索引脚本编写一个独立的脚本用来把data/docs下的文本文件批量导入向量库# 文件路径scripts/index_docs.py import os from app.embeddings import embed_texts from app.vector_store import VectorStore from app.rag import split_text def load_txt_files(docs_dir: str): all_chunks [] all_metas [] for filename in os.listdir(docs_dir): if not filename.endswith(.txt): continue filepath os.path.join(docs_dir, filename) with open(filepath, r, encodingutf-8) as f: content f.read() chunks split_text(content) all_chunks.extend(chunks) all_metas.extend([{source: filename}] * len(chunks)) print(f处理 {filename}: 共 {len(chunks)} 块) return all_chunks, all_metas if __name__ __main__: chunks, metas load_txt_files(data/docs) vs VectorStore() embeddings embed_texts(chunks) vs.collection.add( ids[fdoc_{i} for i in range(len(chunks))], embeddingsembeddings, documentschunks, metadatasmetas, ) print(f索引完成共 {len(chunks)} 个文本块)这里的embed_texts做了批量编码比循环单条编码快很多。4. Agent 开发让大模型学会调用工具RAG 解决了“知识”问题Agent 解决的是“行动”问题。这一节的示例会实现一个简单但完整的 ReAct Agent它能调用两个工具获取天气和计算数学表达式。4.1 工具定义在 Agent 里工具本质上是“一个函数 一份 JSON Schema 描述”。大模型根据描述决定要不要调用、传入什么参数。# 文件路径app/tools.py import math TOOL_SCHEMAS [ { type: function, function: { name: get_weather, description: 查询指定城市的当前天气, parameters: { type: object, properties: { city: { type: string, description: 城市名例如 北京、上海、广州 } }, required: [city] } } }, { type: function, function: { name: calculate, description: 计算数学表达式, parameters: { type: object, properties: { expression: { type: string, description: 数学表达式例如 (12 3) * 4 / 2 } }, required: [expression] } } } ] def get_weather(city: str) - str: # 生产环境这里换成真实天气 API return f{city}晴25℃东南风3级 def calculate(expression: str) - str: allowed set(0123456789-*/() .) if not all(c in allowed for c in expression): return 错误表达式包含不支持的字符 try: result eval(expression, {math: math}, {}) return f{expression} {result} except Exception as e: return f计算失败: {e} TOOL_MAP { get_weather: get_weather, calculate: calculate, }注意eval演示只是为了说明工具封装机制生产环境绝对不要直接eval用户输入。真实的数学计算工具应该使用asteval或numexpr这类安全解析库。4.2 Function Calling 核心循环OpenAI 风格的 Function Calling 流程是发送消息带上工具定义。如果模型返回tool_calls执行对应的本地函数。把函数结果作为一条tool消息返回给模型。重复直到模型不再请求调用工具。# 文件路径app/agent.py from openai import OpenAI from .tools import TOOL_SCHEMAS, TOOL_MAP SYSTEM_PROMPT 你是智能助手可以调用工具。 请根据用户需求选择合适的工具工具返回结果后基于结果组织最终回答。 若无需调用工具直接回答即可。 def run_agent(client: OpenAI, user_message: str, max_iterations: int 5): messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_message}, ] for step in range(max_iterations): response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolsTOOL_SCHEMAS, tool_choiceauto, ) msg response.choices[0].message if not msg.tool_calls: return msg.content print(f[第{step 1}步] 模型请求调用工具: {[tc.function.name for tc in msg.tool_calls]}) # 把模型这条带 tool_calls 的消息追加到对话历史 messages.append(msg) for tool_call in msg.tool_calls: fn_name tool_call.function.name fn_args json.loads(tool_call.function.arguments) result TOOL_MAP[fn_name](**fn_args) messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) return 已达最大迭代次数停止执行。这个循环是 Agent 最核心的部分。后面叠加反思、多步规划、记忆本质上都是在扩充“消息列表”和“工具集合”循环机制不变。4.3 Agent 加 RAG工具化知识检索将第 3 节的 RAG 封装成 Agent 的一个工具就得到了一个简单的 Agentic RAG模型判断用户问题是否需要查资料需要时调用search_knowledge_base工具拿到资料后再回答。# 继续在 tools.py 中添加 from .vector_store import VectorStore from .embeddings import get_embedding_model # RAG 工具函数 _search_tool_schema { type: function, function: { name: search_knowledge_base, description: 从企业内部知识库中检索与问题相关的资料片段, parameters: { type: object, properties: { query: {type: string, description: 检索关键词或问题} }, required: [query] } } } def search_knowledge_base(query: str, vs: VectorStore None) - str: model get_embedding_model() query_vec model.encode(query).tolist() docs, metas vs.search(query_vec, top_k3) return \n\n.join(docs)之后把_search_tool_schema添加到TOOL_SCHEMAS把函数注册到TOOL_MAPAgent 就自动具备了“按需检索”能力。这种方法比“每次提问都强制检索”更省 token模型可以先判断问题是否需要外部知识。5. MCP 实战用标准协议接入工具前面第 4 节的工具调用是“硬编码式”的工具定义在代码里客户端直接调用本地函数。现在改用 MCP让工具变成一个可以独立运行的 Server任何 MCP 客户端都能动态发现和调用它。5.1 使用官方 SDK 实现 MCP Server官方 MCP Python SDK 的包名是mcp。安装pip install mcp下面实现一个基于stdio传输的 MCP Server提供两个工具get_current_time和search_documents。# 文件路径app/mcp_server.py from mcp.server.fastmcp import FastMCP import datetime mcp FastMCP(demo-tools) mcp.tool() def get_current_time() - str: 返回当前系统时间 return datetime.datetime.now().isoformat() mcp.tool() def search_documents(query: str, top_k: int 3) - str: 从本地向量库检索文档片段 # 复用第 3 节的逻辑 from .vector_store import VectorStore from .embeddings import get_embedding_model vs VectorStore() model get_embedding_model() query_vec model.encode(query).tolist() docs, _ vs.search(query_vec, top_ktop_k) return \n\n.join(docs) if __name__ __main__: mcp.run(transportstdio)运行python -m app.mcp_server启动后程序会等待 stdin 输入 MCP 协议消息所以看起来“卡住”是正常的。MCP Server 本身不是给人直接交互的进程而是等待被 MCP Client 拉起。5.2 编写一个 MCP ClientMCP Client 负责连接 Server、获取工具列表、调用工具。下面用mcp官方客户端库实现# 文件路径app/mcp_client.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params StdioServerParameters( commandpython, args[-m, app.mcp_server], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 1. 动态发现工具 tools await session.list_tools() print(可用工具:) for t in tools.tools: print(f - {t.name}: {t.description}) # 2. 调用工具 result await session.call_tool( get_current_time, arguments{} ) print(当前时间:, result.content) if __name__ __main__: asyncio.run(main())5.3 把 MCP 工具接入 Agent有了 MCP Client之前的 Agent 就可以不直接依赖TOOL_MAP而是通过 MCP 动态获取工具。核心改动点Agent 启动时连接 MCP Server把list_tools()返回的 Schema 转成 OpenAI Function Calling 的 tools 参数。模型请求调用某个工具时Agent 把请求转发给 MCP Client 的call_tool。工具执行结果返回给模型继续多轮交互。伪代码思路# 文件路径app/agent_with_mcp.py核心改编示意 from mcp import ClientSession async def run_agent_with_mcp(client: OpenAI, user_message: str, session: ClientSession): mcp_tools await session.list_tools() tools [] tool_map {} for t in mcp_tools.tools: tools.append({ type: function, function: { name: t.name, description: t.description, parameters: t.inputSchema, } }) tool_map[t.name] t messages [{role: user, content: user_message}] response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, tool_choiceauto, ) msg response.choices[0].message if msg.tool_calls: messages.append(msg) for tc in msg.tool_calls: result await session.call_tool(tc.function.name, argumentsjson.loads(tc.function.arguments)) messages.append({ role: tool, tool_call_id: tc.id, content: str(result.content), }) response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, ) return response.choices[0].message.contentMCP 把工具从“应用内部函数”变成了“可独立部署的服务”。你可以用 Python 写一个 Server用 Node.js 写另一个 ServerAgent 通过 MCP 统一调度它们完全不需要关心对方的技术栈。6. 全栈实战FastAPI 集成 RAG、Agent、MCP 并支持 SSE 流式输出前面几节的代码都是独立的模块。这一节把它们组合成一个 Web 服务并实现大模型回答的实时流式渲染。6.1 创建 FastAPI 主应用先搭建基本的应用骨架# 文件路径app/main.py import json from fastapi import FastAPI, HTTPException from pydantic import BaseModel from fastapi.middleware.cors import CORSMiddleware from sse_starlette.sse import EventSourceResponse from openai import OpenAI from .config import get_settings from .rag import ask_rag from .vector_store import VectorStore from .embeddings import get_embedding_model app FastAPI(titleAI Learning Hub API) app.add_middleware( CORSMiddleware, allow_origins[*], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) class ChatRequest(BaseModel): question: str use_rag: bool True stream: bool True class ChatResponse(BaseModel): answer: str sources: list[str] []6.2 流式输出SSE 与 AbortSSEServer-Sent Events是一种基于 HTTP 的单向流式传输协议。大模型逐 token 生成结果时服务端通过 SSE 把 token 分批推给前端。前端用fetch配合ReadableStream实时渲染文字。关键点前端如果中断了连接FastAPI 的异步生成器会收到asyncio.CancelledError这时候需要清理资源。使用sse-starlette的EventSourceResponse可以自动处理心跳和断开。如果使用流式接口前端取消请求时浏览器会发送中断信号但服务端需要配合abort才能停止大模型继续生成。# 文件路径app/main.py添加流式接口 import asyncio def stream_chat(question: str, use_rag: bool): settings get_settings() client OpenAI(base_urlsettings.base_url, api_keysettings.api_key) async def event_generator(): contexts [] if use_rag: vs VectorStore() model get_embedding_model() query_vec model.encode(question).tolist() contexts, _ vs.search(query_vec, top_k3) messages [ {role: system, content: 你是 AI 学习助手。}, {role: user, content: question}, ] if contexts: ctx_text \n\n.join(contexts) messages.insert(0, {role: system, content: f参考资料\n{ctx_text}}) try: stream client.chat.completions.create( modelsettings.model, messagesmessages, streamTrue, ) for chunk in stream: delta chunk.choices[0].delta if delta and delta.content: yield {event: message, data: json.dumps({content: delta.content}, ensure_asciiFalse)} except asyncio.CancelledError: # 客户端断开连接提前终止 raise finally: # 清理资源 try: stream.close() except Exception: pass return EventSourceResponse(event_generator()) app.post(/v1/chat/stream) async def chat_stream(req: ChatRequest): if not req.stream: # 非流式模式 client OpenAI(base_urlget_settings().base_url, api_keyget_settings().api_key) response client.chat.completions.create( modelget_settings().model, messages[{role: user, content: req.question}], ) return {answer: response.choices[0].message.content} return stream_chat(req.question, req.use_rag)6.3 启动服务与前端快速测试启动命令uvicorn app.main:app --reload --port 8000前端最简单的测试页面!DOCTYPE html html langzh head meta charsetUTF-8 titleAI 流式问答 Demo/title /head body div idoutput/div input idquestion stylewidth: 400px; placeholder输入问题 button onclicksend()发送/button script async function send() { const question document.getElementById(question).value; const output document.getElementById(output); output.innerText ; const resp await fetch(/v1/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question, use_rag: true, stream: true }) }); const reader resp.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // 解析 SSE 格式 let lines buffer.split(\n); buffer lines.pop(); for (const line of lines) { if (line.startsWith(data: )) { try { const json JSON.parse(line.slice(6)); output.innerText json.content; } catch (e) {} } } } } /script /body /html前端使用AbortController实现取消操作let controller null; async function send() { if (controller) controller.abort(); controller new AbortController(); const resp await fetch(/v1/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question, use_rag: true, stream: true }), signal: controller.signal }); // 后续读取逻辑同上 }为什么前端要单独处理abort因为大模型流式生成很消耗服务端资源。如果用户在答案生成到一半时切换问题前端必须立刻断开旧连接服务端才会收到CancelledError并停止生成。否则旧的请求还会在服务器上继续跑浪费 token 和 GPU 资源。6.4 集成 MCP Server 到 FastAPI 生命周期在应用启动时拉起 MCP Server在关闭时优雅退出# 文件路径app/main.py补充生命周期 import subprocess import sys mcp_process None app.on_event(startup) def startup(): global mcp_process # 以子进程方式启动 MCP Server mcp_process subprocess.Popen( [sys.executable, -m, app.mcp_server], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, ) print(MCP Server 已启动) app.on_event(shutdown) def shutdown(): global mcp_process if mcp_process: mcp_process.terminate() mcp_process.wait() print(MCP Server 已停止)更规范的做法是通过stdio_client建立异步会话将会话对象挂到应用 state 上后续请求直接使用。上面的子进程启动方式适合快速演示生产环境建议用 supervisord 或 docker-compose 统一管理进程。7. 常见问题与排查清单7.1 RAG 相关问题问题现象常见原因解决思路检索结果不相关嵌入模型不适合领域文本换用领域微调嵌入模型检查切块是否太大/太小检索到很多垃圾片段元数据没有过滤检索时增加 metadata 过滤条件如限定文档分类回答仍然出现幻觉参考片段没有真正参与生成打印实际 prompt确认检索片段是否被正确拼入向量库导入速度慢逐条 encode改为批量嵌入用encode(texts)一次处理多条Chroma 数据冲突ID 重复使用带 hash 的唯一 ID或用 collection.upsert7.2 Agent 相关问题问题现象常见原因解决思路模型不调用工具tool schema 描述不清晰检查description是否明确将tool_choice设为auto或required工具参数解析报错模型返回非法 JSON尝试用json.loads解析并捕获异常如频繁失败将参数定义更简单Agent 陷入死循环缺少迭代上限设置max_iterations达到上限返回当前已执行结果多次重复同一工具没有记录历史消息确保每次循环把 assistant 消息和 tool 消息追加到 messages工具执行报错后 Agent 放弃没有错误反馈机制工具结果即使失败也要返回错误信息字符串让模型重新决策7.3 MCP 相关问题问题现象常见原因解决思路MCP Server 启动后立即退出配置错误或依赖缺失单独运行python -m app.mcp_server查看控制台报错Client 连不上 Servertransport 不匹配stdio 需要手动拉起进程root URL 需要 server 运行在 HTTP 模式工具方法返回值异常返回了不可序列化对象MCP 要求返回字符串或结构化 JSON 可序列化内容动态发现不到工具装饰器未生效确保工具函数写在FastMCP(name)实例之后并使用了mcp.tool()跨平台换行问题Windows 下路径分隔符使用pathlib.Path, 不要硬编码路径7.4 通用排查步骤遇到问题, 我建议按这个顺序排查最小化复现删掉无关代码只保留能触发问题的最小脚本。打印输入输出把 prompt、检索结果、工具返回结果都打印出来很多问题一眼就能看出来。检查网络日志查看是否有超时、认证失败、速率限制。对比官方示例如果改了某个库的用法先回退到官方示例验证环境是否正常。看版本更新库的 API 变化频繁出现不认识的参数先去官网看 Changelog。8. 最佳实践与工程建议8.1 关于 RAG 的最佳实践先做文档体检再切块。 如果原始文档里大量表格、图片直接纯文本切块会丢失结构。建议在切块前先用 Markdown 或 HTML 格式做清洗保留标题层级。标题信息也可以写入 metadata比如{h1: ..., h2: ...}检索时优先匹配标题相似的片段。检索打分不能只看相似度。 生产环境建议在 RAG 链路中加入重排序Reranking。先用向量检索召回 top 20再用交叉编码器重排取 top 3-5。这能显著提升准确率但会带来额外延迟。知识库几千条以内的简单场景可以暂不做重排。评估比调参重要。 建立一套评估数据集比如 30 条真实业务问题每条标明期望答案片段。每次改索引策略后在这套数据集上计算hit_rate和answer_relevancy用数据指导优化不要靠感觉调 chunk_size。考虑增量更新。 生产环境不能每次全量重建向量库。文档新增时只对新增文档做切块和嵌入文档删除时根据 metadata 中的 source 字段删除对应向量。Chroma 支持条件删除但如果删除条件复杂推荐维护一份“文档哈希 ↔ 向量 ID”的映射表。8.2 关于 Agent 的最佳实践工具数量控制。 把 20 个工具暴露给模型效果不一定比 5 个好。模型工具选择的准确率会随着工具数量增加而下降。建议优先暴露核心工具长尾工具包成“执行指定 id 的任务”这种通用接口。任务拆解状态可视化。 在复杂 Agent 场景中一定要把“当前正在执行哪一步”“下一步计划是什么”通过 UI 或日志输出。用户看到 Agent 在干什么信任度会大幅提升。SSE 可以不仅推送模型 token也可以推送 agent 中间的思考记录。安全是大前提。 给 Agent 接入数据库、文件系统、支付这类敏感工具时务必加上权限控制层。建议设计一组最小权限函数即使是工具函数内部也要做参数校验并且明确日志记录谁在什么时候调用了什么工具。涉及删除、覆盖文件等危险操作时强制二次确认。8.3 关于 MCP 的最佳实践一个能力一个 Server。 MCP Server 不要把所有工具都塞进去。按照领域拆分比如database-server、search-server、git-server。每个 Server 独立版本管理独立发布Agent 按需加载。鉴权与隔离。 企业内部 MCP Server 需要鉴权。stdio 模式跑在本机相对安全但 HTTP/SSE 模式暴露到网络时必须加 token 校验。不要将内网 MCP Server 不设防地绑定到公网端口。先协议后框架。 MCP 的核心是协议不是 SDK。官方 SDK 会持续迭代但协议本身的工具列表、调用返回、资源模型变动较小。遇到 SDK 问题先用 curl 测试协议是否符合预期再排查代码。8.4 关于流式输出的工程建议区分业务流与模型流。 SSE 事件不要全都叫message。建议定义多种事件类型event: status表示“正在检索文档”“正在调用工具”。event: message表示模型真实输出 token。event: done表示本次请求结束。前端按事件类型分别渲染体验会好很多。使用请求 ID 追踪全链路。 每个请求生成一个 UUID在日志、指标采集、异常上报里带上它。流式场景中排查问题会更困难请求 ID 是唯一的线索。超时管理。 大模型接口可能长时间不返回。OpenAI Python SDK 有默认超时但流式接口需要单独设置timeout和max_retries。长任务建议使用任务队列异步处理而不是让一个 HTTP 连接等十分钟。9. 结语从 RAG 到 Agent 再到 MCP这三块内容构成了 2026 年大模型应用开发的主线能力。文章里的所有代码都可以直接复制修改使用建议你按以下顺序动手实验第一次运行搭建 FastAPI 服务调用普通非流式接口。第二次运行导入几篇真实文档测试 RAG 检索效果。第三次运行让 Agent 通过 MCP 调用工具观察它如何规划步骤。第四次运行接上流式输出在前端页面观察逐字渲染效果。整个链路跑通之后再回到业务中思考你的场景最需要的是知识问答RAG还是自动化任务Agent还是统一工具接入MCP不要一开始就把所有功能都做进去。一个好的大模型应用往往是从一个最小闭环开始逐步扩展能力边界。如果这篇文章对你有用欢迎收藏备用。下一篇我会继续拆解 RAG 评估指标、重排序算法选型以及 Agent 记忆组件的设计。有任何问题可以在评论区留言我会挑有代表性的问题展开讲。 RAG、Agent与MCP实战：大模型应用开发核心主线
之前带过不少想入门 AI 大模型应用开发的朋友发现大家最常遇到的问题并不是“看不懂论文”而是不知道从哪里开始动手。网上资料要么是纯概念科普看完还是不会写代码要么是某个框架的碎片教程换个场景就抓瞎。这篇文章我打算把 AI 大模型应用开发最核心的三块内容——RAG、Agent、MCP串成一条完整的学习主线从零开始讲原理、写代码、跑通一个可演示的项目。无论你是刚接触 Python 的后端开发、想转 AI 应用方向的学生还是已经在业务里尝试接入大模型的工程师这篇文章都能帮你建立一套相对完整的知识框架。以下内容会按照“概念 → 环境 → RAG → Agent → MCP → 全栈实战 → 排错 → 最佳实践”的顺序展开。代码以 Python 为主会给出完整的可运行示例。所有项目会尽量保持模块化方便你根据自己的业务场景裁剪使用。1. 为什么要同时学 RAG、Agent 和 MCP很多人会把 RAG、Agent、MCP 混在一起聊但它们其实是三个不同层面的问题。只有把三者分开理解再组合起来才能真正明白大模型应用开发的完整路径。1.1 RAG解决模型“不知道”的问题大模型在训练完成之后知识就固定了。这里有两个致命问题知识滞后模型不会自动知道昨天新发布的产品文档。幻觉问题模型对自己不知道的内容可能一本正经地用不存在的知识回答你。RAGRetrieval-Augmented Generation检索增强生成的思路是提问之前先从外部知识库里检索出和问题相关的片段把片段拼进提示词再让大模型基于这些片段作答。相当于给模型开卷考试而不是闭卷瞎编。RAG 的核心流程可以简化成五步把文档切分成小块chunk。用嵌入模型Embedding Model把每一块转成向量。把向量和文本一起存进向量数据库。用户提问时把问题也转成向量做相似度检索。把检索结果和问题一起发给大模型生成最终答案。1.2 Agent解决模型“不能做”的问题普通聊天只能“说”Agent智能体的意义在于“做”。一个 Agent 通常包含以下能力规划Planning把复杂任务拆成多个步骤。工具调用Tool Use通过函数调用获取数据、操作系统、读写文件、查询数据库。记忆Memory在多次交互中记住上下文。反思Reflection执行失败后重新决定下一步行动。比较经典的 Agent 工作流是 ReAct模型先推理Reasoning出当前该做什么然后调用工具Acting拿到工具结果后再继续推理直到任务完成。1.3 MCP解决工具接入标准问题当你想让大模型调用工具时最原始的做法是“写死在代码里”def call_tool(name, args): if name get_weather: return weather_api(args[city])这种方式在只有两三个工具时很清晰但在复杂项目里会出现问题每个工具都要单独封装、协议不统一、权限模型混乱、调试困难。MCPModel Context Protocol模型上下文协议就是来解决这个问题的标准化协议。MCP 的架构可以类比成“AI 应用的 USB-C 接口”MCP Server提供工具、资源或提示词比如“数据库查询工具”“GitHub 操作工具”。MCP Client连接 Server 的一方通常运行在大模型应用侧。工具发现机制客户端可以动态获取 Server 支持的工具列表、参数结构不再硬编码。一个标准的 MCP Server 可以通过stdio或sse传输方式运行。2026 年的今天MCP 已经不只是实验性概念不少浏览器扩展、调试工具、数据库客户端都开始提供 MCP Server 接口。这也是为什么现在 AI 应用开发面试里MCP 几乎是必问题。1.4 三者的关系可以把三者的关系总结成一句话RAG 管知识Agent 管决策与行动MCP 管工具接入标准。一个完整的大模型应用通常是“RAG 提供知识基础 → Agent 负责拆解任务 → MCP 负责对接各种工具”。后面第 5 章的实战项目会把三者组合在一起演示。2. 环境准备与学习路线2.1 版本与工具说明由于大模型相关库迭代速度非常快我不建议给出具体的硬编码版本号。本文示例以以下环境为基准但你实际使用时需要根据项目情况调整组件推荐选择说明操作系统Windows 10/11、macOS、Linux 均可本文代码是跨平台的Windows 使用python -m venv创建虚拟环境Python3.10 以上推荐 3.11 或 3.12对类型注解和异步支持更好包管理pip 或 uv建议先掌握 pipuv 后续再学IDEVS Code / PyCharm推荐 VS Code Python 插件大模型 APIOpenAI 兼容接口国内厂商、本地模型Ollama/vLLM大多提供 OpenAI 兼容接口向量数据库Chroma本地开发/ FAISS / Milvus生产本文先用 Chroma零配置最容易跑通嵌入模型本地 sentence-transformers 或 API 嵌入演示用本地小模型避免额外费用2.2 依赖安装创建一个新的项目目录并初始化虚拟环境mkdir ai-learning-hub cd ai-learning-hub python -m venv venvWindows 激活虚拟环境venv\Scripts\activatemacOS / Linuxsource venv/bin/activate安装基础依赖pip install fastapi uvicorn openai chromadb langchain-core langchain-community python-dotenv httpx sse-starlette这些库的作用fastapi和uvicorn用于搭建 Web 服务。openai调用 OpenAI 兼容接口。chromadb本地向量数据库。langchain-core和langchain-community提供链式抽象和文档加载工具。httpx异步 HTTP 客户端后续 Agent 调用工具用。sse-starlette实现 SSE 流式输出。2.3 项目总结构整个实战项目我建议按下面结构组织ai-learning-hub/ ├── .env ├── requirements.txt ├── app/ │ ├── __init__.py │ ├── config.py │ ├── llm.py │ ├── embeddings.py │ ├── vector_store.py │ ├── rag.py │ ├── tools.py │ ├── agent.py │ ├── mcp_server.py │ ├── mcp_client.py │ └── main.py ├── data/ │ └── docs/ # 放你的知识库文档 └── tests/ └── test_rag.py后面的代码会按这个路径组织。为了便于理解每一小节会给出文件路径方便你对照建文件。3. RAG 实战从零搭建一个本地知识库问答系统这一节先不依赖 LangChain而是手动实现一个最小 RAG 流程。只有理解了每一步在做什么后面用框架时才知道哪里出了问题。3.1 文档切块Chunking为什么要切块因为大模型有上下文窗口限制而且把整篇几万字的文档塞进去效果不一定好。检索时我们希望找到与问题最相关的那一两段而不是整篇文章。切块的策略有很多最简单的是按固定字数切并且上下两块之间保留重叠区域。# 文件路径app/rag.py部分代码 from typing import List def split_text(text: str, chunk_size: int 500, overlap: int 50) - List[str]: 将长文本按 chunk_size 切块相邻块之间保留 overlap 字符重叠。 重叠可以避免关键信息恰好落在两个块的边界。 if len(text) chunk_size: return [text] chunks [] start 0 while start len(text): end min(start chunk_size, len(text)) # 尽量在段落或句子边界处切分 if end len(text): # 从 end 往前找最近的换行符 last_newline text.rfind(\n, start, end) if last_newline ! -1 and end - last_newline 100: end last_newline 1 chunks.append(text[start:end]) start max(end - overlap, 1) return chunks这里有一个关键点切块大小没有绝对标准。如果业务文档是 FAQ 类可以直接按条目切如果是长篇技术文档500 字左右通常比较合适。块太大检索出来容易超过上下文窗口块太小语义不完整检索准确率会下降。3.2 嵌入与向量化嵌入Embedding是把文本变成一组浮点数向量的过程。语义接近的两个句子它们在向量空间中的距离也接近。这里使用 sentence-transformers 加载本地小模型# 文件路径app/embeddings.py from sentence_transformers import SentenceTransformer _model None def get_embedding_model(): global _model if _model is None: # 这是一个轻量级中文嵌入模型首次运行会自动下载 _model SentenceTransformer(shibing624/text2vec-base-chinese) return _model def embed_text(text: str): model get_embedding_model() return model.encode(text).tolist() def embed_texts(texts): model get_embedding_model() return model.encode(texts).tolist()注意直接用pip install sentence-transformers安装即可。如果你的环境无法访问 Hugging Face比如网络策略限制可以把模型文件提前下载到本地然后通过SentenceTransformer(本地路径)加载。本地加载方式适合企业内部知识库场景。3.3 向量存储与检索Chroma 是一个非常好用的本地向量数据库。它会自动把向量和文档内容持久化在本地目录。# 文件路径app/vector_store.py import chromadb from chromadb.config import Settings class VectorStore: def __init__(self, persist_dir./chroma_db, collection_namedocs): self.client chromadb.PersistentClient( pathpersist_dir, settingsSettings(anonymized_telemetryFalse) ) self.collection self.client.get_or_create_collection( namecollection_name, metadata{hnsw:space: cosine} # 使用余弦距离 ) def add_documents(self, docs: list[str], metadatas: list[dict] | None None): ids [fdoc_{i}_{hash(doc)} for i, doc in enumerate(docs)] self.collection.add( idsids, documentsdocs, metadatasmetadatas or [{}] * len(docs) ) def search(self, query_embedding: list[float], top_k: int 3): result self.collection.query( query_embeddings[query_embedding], n_resultstop_k, ) return result[documents][0], result[metadatas][0]有几个容易踩坑的地方距离函数文本检索一般用余弦距离cosine不要选默认的 L2。ID 唯一性相同文档重复插入时 ID 不能冲突否则会覆盖或被拒绝。metadata 数量metadatas数量必须和documents一致。3.4 构建 prompt 并调用大模型检索到相关片段后下一步是构造 prompt。这里用一个简单的模板# 文件路径app/rag.py接前面 from openai import OpenAI def build_prompt(question: str, contexts: list[str]) - str: context_text \n\n.join([f[资料{i1}]\n{ctx} for i, ctx in enumerate(contexts)]) return f你是企业内部知识助手请严格依据参考资料回答问题。 如果参考资料中找不到答案请直接说明“资料库中暂无相关内容”不要编造。 参考资料 {context_text} 用户问题{question} 请用简洁、准确的中文回答 def ask_rag(question: str, vector_store: VectorStore, embedding_model, client: OpenAI, top_k: int 3): # 1. 把问题转成向量 query_vec embedding_model.encode(question).tolist() # 2. 检索 docs, metadatas vector_store.search(query_vec, top_ktop_k) # 3. 构造 prompt prompt build_prompt(question, docs) # 4. 调用大模型 response client.chat.completions.create( modelgpt-4o-mini, # 根据你的模型服务商调整 messages[{role: user, content: prompt}], temperature0.3, ) return response.choices[0].message.content, docs如果是本地部署的大模型只需要修改base_url和model。比如使用 Ollamaclient OpenAI( base_urlhttp://localhost:11434/v1, api_keyollama, # Ollama 本地不需要真实 key )把model换成你本地拉取的模型名比如qwen2.5:7b代码其他部分完全不用改。这就是 OpenAI 兼容接口的价值。3.5 本地知识库索引脚本编写一个独立的脚本用来把data/docs下的文本文件批量导入向量库# 文件路径scripts/index_docs.py import os from app.embeddings import embed_texts from app.vector_store import VectorStore from app.rag import split_text def load_txt_files(docs_dir: str): all_chunks [] all_metas [] for filename in os.listdir(docs_dir): if not filename.endswith(.txt): continue filepath os.path.join(docs_dir, filename) with open(filepath, r, encodingutf-8) as f: content f.read() chunks split_text(content) all_chunks.extend(chunks) all_metas.extend([{source: filename}] * len(chunks)) print(f处理 {filename}: 共 {len(chunks)} 块) return all_chunks, all_metas if __name__ __main__: chunks, metas load_txt_files(data/docs) vs VectorStore() embeddings embed_texts(chunks) vs.collection.add( ids[fdoc_{i} for i in range(len(chunks))], embeddingsembeddings, documentschunks, metadatasmetas, ) print(f索引完成共 {len(chunks)} 个文本块)这里的embed_texts做了批量编码比循环单条编码快很多。4. Agent 开发让大模型学会调用工具RAG 解决了“知识”问题Agent 解决的是“行动”问题。这一节的示例会实现一个简单但完整的 ReAct Agent它能调用两个工具获取天气和计算数学表达式。4.1 工具定义在 Agent 里工具本质上是“一个函数 一份 JSON Schema 描述”。大模型根据描述决定要不要调用、传入什么参数。# 文件路径app/tools.py import math TOOL_SCHEMAS [ { type: function, function: { name: get_weather, description: 查询指定城市的当前天气, parameters: { type: object, properties: { city: { type: string, description: 城市名例如 北京、上海、广州 } }, required: [city] } } }, { type: function, function: { name: calculate, description: 计算数学表达式, parameters: { type: object, properties: { expression: { type: string, description: 数学表达式例如 (12 3) * 4 / 2 } }, required: [expression] } } } ] def get_weather(city: str) - str: # 生产环境这里换成真实天气 API return f{city}晴25℃东南风3级 def calculate(expression: str) - str: allowed set(0123456789-*/() .) if not all(c in allowed for c in expression): return 错误表达式包含不支持的字符 try: result eval(expression, {math: math}, {}) return f{expression} {result} except Exception as e: return f计算失败: {e} TOOL_MAP { get_weather: get_weather, calculate: calculate, }注意eval演示只是为了说明工具封装机制生产环境绝对不要直接eval用户输入。真实的数学计算工具应该使用asteval或numexpr这类安全解析库。4.2 Function Calling 核心循环OpenAI 风格的 Function Calling 流程是发送消息带上工具定义。如果模型返回tool_calls执行对应的本地函数。把函数结果作为一条tool消息返回给模型。重复直到模型不再请求调用工具。# 文件路径app/agent.py from openai import OpenAI from .tools import TOOL_SCHEMAS, TOOL_MAP SYSTEM_PROMPT 你是智能助手可以调用工具。 请根据用户需求选择合适的工具工具返回结果后基于结果组织最终回答。 若无需调用工具直接回答即可。 def run_agent(client: OpenAI, user_message: str, max_iterations: int 5): messages [ {role: system, content: SYSTEM_PROMPT}, {role: user, content: user_message}, ] for step in range(max_iterations): response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolsTOOL_SCHEMAS, tool_choiceauto, ) msg response.choices[0].message if not msg.tool_calls: return msg.content print(f[第{step 1}步] 模型请求调用工具: {[tc.function.name for tc in msg.tool_calls]}) # 把模型这条带 tool_calls 的消息追加到对话历史 messages.append(msg) for tool_call in msg.tool_calls: fn_name tool_call.function.name fn_args json.loads(tool_call.function.arguments) result TOOL_MAP[fn_name](**fn_args) messages.append({ role: tool, tool_call_id: tool_call.id, content: result, }) return 已达最大迭代次数停止执行。这个循环是 Agent 最核心的部分。后面叠加反思、多步规划、记忆本质上都是在扩充“消息列表”和“工具集合”循环机制不变。4.3 Agent 加 RAG工具化知识检索将第 3 节的 RAG 封装成 Agent 的一个工具就得到了一个简单的 Agentic RAG模型判断用户问题是否需要查资料需要时调用search_knowledge_base工具拿到资料后再回答。# 继续在 tools.py 中添加 from .vector_store import VectorStore from .embeddings import get_embedding_model # RAG 工具函数 _search_tool_schema { type: function, function: { name: search_knowledge_base, description: 从企业内部知识库中检索与问题相关的资料片段, parameters: { type: object, properties: { query: {type: string, description: 检索关键词或问题} }, required: [query] } } } def search_knowledge_base(query: str, vs: VectorStore None) - str: model get_embedding_model() query_vec model.encode(query).tolist() docs, metas vs.search(query_vec, top_k3) return \n\n.join(docs)之后把_search_tool_schema添加到TOOL_SCHEMAS把函数注册到TOOL_MAPAgent 就自动具备了“按需检索”能力。这种方法比“每次提问都强制检索”更省 token模型可以先判断问题是否需要外部知识。5. MCP 实战用标准协议接入工具前面第 4 节的工具调用是“硬编码式”的工具定义在代码里客户端直接调用本地函数。现在改用 MCP让工具变成一个可以独立运行的 Server任何 MCP 客户端都能动态发现和调用它。5.1 使用官方 SDK 实现 MCP Server官方 MCP Python SDK 的包名是mcp。安装pip install mcp下面实现一个基于stdio传输的 MCP Server提供两个工具get_current_time和search_documents。# 文件路径app/mcp_server.py from mcp.server.fastmcp import FastMCP import datetime mcp FastMCP(demo-tools) mcp.tool() def get_current_time() - str: 返回当前系统时间 return datetime.datetime.now().isoformat() mcp.tool() def search_documents(query: str, top_k: int 3) - str: 从本地向量库检索文档片段 # 复用第 3 节的逻辑 from .vector_store import VectorStore from .embeddings import get_embedding_model vs VectorStore() model get_embedding_model() query_vec model.encode(query).tolist() docs, _ vs.search(query_vec, top_ktop_k) return \n\n.join(docs) if __name__ __main__: mcp.run(transportstdio)运行python -m app.mcp_server启动后程序会等待 stdin 输入 MCP 协议消息所以看起来“卡住”是正常的。MCP Server 本身不是给人直接交互的进程而是等待被 MCP Client 拉起。5.2 编写一个 MCP ClientMCP Client 负责连接 Server、获取工具列表、调用工具。下面用mcp官方客户端库实现# 文件路径app/mcp_client.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params StdioServerParameters( commandpython, args[-m, app.mcp_server], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 1. 动态发现工具 tools await session.list_tools() print(可用工具:) for t in tools.tools: print(f - {t.name}: {t.description}) # 2. 调用工具 result await session.call_tool( get_current_time, arguments{} ) print(当前时间:, result.content) if __name__ __main__: asyncio.run(main())5.3 把 MCP 工具接入 Agent有了 MCP Client之前的 Agent 就可以不直接依赖TOOL_MAP而是通过 MCP 动态获取工具。核心改动点Agent 启动时连接 MCP Server把list_tools()返回的 Schema 转成 OpenAI Function Calling 的 tools 参数。模型请求调用某个工具时Agent 把请求转发给 MCP Client 的call_tool。工具执行结果返回给模型继续多轮交互。伪代码思路# 文件路径app/agent_with_mcp.py核心改编示意 from mcp import ClientSession async def run_agent_with_mcp(client: OpenAI, user_message: str, session: ClientSession): mcp_tools await session.list_tools() tools [] tool_map {} for t in mcp_tools.tools: tools.append({ type: function, function: { name: t.name, description: t.description, parameters: t.inputSchema, } }) tool_map[t.name] t messages [{role: user, content: user_message}] response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, tool_choiceauto, ) msg response.choices[0].message if msg.tool_calls: messages.append(msg) for tc in msg.tool_calls: result await session.call_tool(tc.function.name, argumentsjson.loads(tc.function.arguments)) messages.append({ role: tool, tool_call_id: tc.id, content: str(result.content), }) response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, ) return response.choices[0].message.contentMCP 把工具从“应用内部函数”变成了“可独立部署的服务”。你可以用 Python 写一个 Server用 Node.js 写另一个 ServerAgent 通过 MCP 统一调度它们完全不需要关心对方的技术栈。6. 全栈实战FastAPI 集成 RAG、Agent、MCP 并支持 SSE 流式输出前面几节的代码都是独立的模块。这一节把它们组合成一个 Web 服务并实现大模型回答的实时流式渲染。6.1 创建 FastAPI 主应用先搭建基本的应用骨架# 文件路径app/main.py import json from fastapi import FastAPI, HTTPException from pydantic import BaseModel from fastapi.middleware.cors import CORSMiddleware from sse_starlette.sse import EventSourceResponse from openai import OpenAI from .config import get_settings from .rag import ask_rag from .vector_store import VectorStore from .embeddings import get_embedding_model app FastAPI(titleAI Learning Hub API) app.add_middleware( CORSMiddleware, allow_origins[*], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) class ChatRequest(BaseModel): question: str use_rag: bool True stream: bool True class ChatResponse(BaseModel): answer: str sources: list[str] []6.2 流式输出SSE 与 AbortSSEServer-Sent Events是一种基于 HTTP 的单向流式传输协议。大模型逐 token 生成结果时服务端通过 SSE 把 token 分批推给前端。前端用fetch配合ReadableStream实时渲染文字。关键点前端如果中断了连接FastAPI 的异步生成器会收到asyncio.CancelledError这时候需要清理资源。使用sse-starlette的EventSourceResponse可以自动处理心跳和断开。如果使用流式接口前端取消请求时浏览器会发送中断信号但服务端需要配合abort才能停止大模型继续生成。# 文件路径app/main.py添加流式接口 import asyncio def stream_chat(question: str, use_rag: bool): settings get_settings() client OpenAI(base_urlsettings.base_url, api_keysettings.api_key) async def event_generator(): contexts [] if use_rag: vs VectorStore() model get_embedding_model() query_vec model.encode(question).tolist() contexts, _ vs.search(query_vec, top_k3) messages [ {role: system, content: 你是 AI 学习助手。}, {role: user, content: question}, ] if contexts: ctx_text \n\n.join(contexts) messages.insert(0, {role: system, content: f参考资料\n{ctx_text}}) try: stream client.chat.completions.create( modelsettings.model, messagesmessages, streamTrue, ) for chunk in stream: delta chunk.choices[0].delta if delta and delta.content: yield {event: message, data: json.dumps({content: delta.content}, ensure_asciiFalse)} except asyncio.CancelledError: # 客户端断开连接提前终止 raise finally: # 清理资源 try: stream.close() except Exception: pass return EventSourceResponse(event_generator()) app.post(/v1/chat/stream) async def chat_stream(req: ChatRequest): if not req.stream: # 非流式模式 client OpenAI(base_urlget_settings().base_url, api_keyget_settings().api_key) response client.chat.completions.create( modelget_settings().model, messages[{role: user, content: req.question}], ) return {answer: response.choices[0].message.content} return stream_chat(req.question, req.use_rag)6.3 启动服务与前端快速测试启动命令uvicorn app.main:app --reload --port 8000前端最简单的测试页面!DOCTYPE html html langzh head meta charsetUTF-8 titleAI 流式问答 Demo/title /head body div idoutput/div input idquestion stylewidth: 400px; placeholder输入问题 button onclicksend()发送/button script async function send() { const question document.getElementById(question).value; const output document.getElementById(output); output.innerText ; const resp await fetch(/v1/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question, use_rag: true, stream: true }) }); const reader resp.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // 解析 SSE 格式 let lines buffer.split(\n); buffer lines.pop(); for (const line of lines) { if (line.startsWith(data: )) { try { const json JSON.parse(line.slice(6)); output.innerText json.content; } catch (e) {} } } } } /script /body /html前端使用AbortController实现取消操作let controller null; async function send() { if (controller) controller.abort(); controller new AbortController(); const resp await fetch(/v1/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question, use_rag: true, stream: true }), signal: controller.signal }); // 后续读取逻辑同上 }为什么前端要单独处理abort因为大模型流式生成很消耗服务端资源。如果用户在答案生成到一半时切换问题前端必须立刻断开旧连接服务端才会收到CancelledError并停止生成。否则旧的请求还会在服务器上继续跑浪费 token 和 GPU 资源。6.4 集成 MCP Server 到 FastAPI 生命周期在应用启动时拉起 MCP Server在关闭时优雅退出# 文件路径app/main.py补充生命周期 import subprocess import sys mcp_process None app.on_event(startup) def startup(): global mcp_process # 以子进程方式启动 MCP Server mcp_process subprocess.Popen( [sys.executable, -m, app.mcp_server], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, ) print(MCP Server 已启动) app.on_event(shutdown) def shutdown(): global mcp_process if mcp_process: mcp_process.terminate() mcp_process.wait() print(MCP Server 已停止)更规范的做法是通过stdio_client建立异步会话将会话对象挂到应用 state 上后续请求直接使用。上面的子进程启动方式适合快速演示生产环境建议用 supervisord 或 docker-compose 统一管理进程。7. 常见问题与排查清单7.1 RAG 相关问题问题现象常见原因解决思路检索结果不相关嵌入模型不适合领域文本换用领域微调嵌入模型检查切块是否太大/太小检索到很多垃圾片段元数据没有过滤检索时增加 metadata 过滤条件如限定文档分类回答仍然出现幻觉参考片段没有真正参与生成打印实际 prompt确认检索片段是否被正确拼入向量库导入速度慢逐条 encode改为批量嵌入用encode(texts)一次处理多条Chroma 数据冲突ID 重复使用带 hash 的唯一 ID或用 collection.upsert7.2 Agent 相关问题问题现象常见原因解决思路模型不调用工具tool schema 描述不清晰检查description是否明确将tool_choice设为auto或required工具参数解析报错模型返回非法 JSON尝试用json.loads解析并捕获异常如频繁失败将参数定义更简单Agent 陷入死循环缺少迭代上限设置max_iterations达到上限返回当前已执行结果多次重复同一工具没有记录历史消息确保每次循环把 assistant 消息和 tool 消息追加到 messages工具执行报错后 Agent 放弃没有错误反馈机制工具结果即使失败也要返回错误信息字符串让模型重新决策7.3 MCP 相关问题问题现象常见原因解决思路MCP Server 启动后立即退出配置错误或依赖缺失单独运行python -m app.mcp_server查看控制台报错Client 连不上 Servertransport 不匹配stdio 需要手动拉起进程root URL 需要 server 运行在 HTTP 模式工具方法返回值异常返回了不可序列化对象MCP 要求返回字符串或结构化 JSON 可序列化内容动态发现不到工具装饰器未生效确保工具函数写在FastMCP(name)实例之后并使用了mcp.tool()跨平台换行问题Windows 下路径分隔符使用pathlib.Path, 不要硬编码路径7.4 通用排查步骤遇到问题, 我建议按这个顺序排查最小化复现删掉无关代码只保留能触发问题的最小脚本。打印输入输出把 prompt、检索结果、工具返回结果都打印出来很多问题一眼就能看出来。检查网络日志查看是否有超时、认证失败、速率限制。对比官方示例如果改了某个库的用法先回退到官方示例验证环境是否正常。看版本更新库的 API 变化频繁出现不认识的参数先去官网看 Changelog。8. 最佳实践与工程建议8.1 关于 RAG 的最佳实践先做文档体检再切块。 如果原始文档里大量表格、图片直接纯文本切块会丢失结构。建议在切块前先用 Markdown 或 HTML 格式做清洗保留标题层级。标题信息也可以写入 metadata比如{h1: ..., h2: ...}检索时优先匹配标题相似的片段。检索打分不能只看相似度。 生产环境建议在 RAG 链路中加入重排序Reranking。先用向量检索召回 top 20再用交叉编码器重排取 top 3-5。这能显著提升准确率但会带来额外延迟。知识库几千条以内的简单场景可以暂不做重排。评估比调参重要。 建立一套评估数据集比如 30 条真实业务问题每条标明期望答案片段。每次改索引策略后在这套数据集上计算hit_rate和answer_relevancy用数据指导优化不要靠感觉调 chunk_size。考虑增量更新。 生产环境不能每次全量重建向量库。文档新增时只对新增文档做切块和嵌入文档删除时根据 metadata 中的 source 字段删除对应向量。Chroma 支持条件删除但如果删除条件复杂推荐维护一份“文档哈希 ↔ 向量 ID”的映射表。8.2 关于 Agent 的最佳实践工具数量控制。 把 20 个工具暴露给模型效果不一定比 5 个好。模型工具选择的准确率会随着工具数量增加而下降。建议优先暴露核心工具长尾工具包成“执行指定 id 的任务”这种通用接口。任务拆解状态可视化。 在复杂 Agent 场景中一定要把“当前正在执行哪一步”“下一步计划是什么”通过 UI 或日志输出。用户看到 Agent 在干什么信任度会大幅提升。SSE 可以不仅推送模型 token也可以推送 agent 中间的思考记录。安全是大前提。 给 Agent 接入数据库、文件系统、支付这类敏感工具时务必加上权限控制层。建议设计一组最小权限函数即使是工具函数内部也要做参数校验并且明确日志记录谁在什么时候调用了什么工具。涉及删除、覆盖文件等危险操作时强制二次确认。8.3 关于 MCP 的最佳实践一个能力一个 Server。 MCP Server 不要把所有工具都塞进去。按照领域拆分比如database-server、search-server、git-server。每个 Server 独立版本管理独立发布Agent 按需加载。鉴权与隔离。 企业内部 MCP Server 需要鉴权。stdio 模式跑在本机相对安全但 HTTP/SSE 模式暴露到网络时必须加 token 校验。不要将内网 MCP Server 不设防地绑定到公网端口。先协议后框架。 MCP 的核心是协议不是 SDK。官方 SDK 会持续迭代但协议本身的工具列表、调用返回、资源模型变动较小。遇到 SDK 问题先用 curl 测试协议是否符合预期再排查代码。8.4 关于流式输出的工程建议区分业务流与模型流。 SSE 事件不要全都叫message。建议定义多种事件类型event: status表示“正在检索文档”“正在调用工具”。event: message表示模型真实输出 token。event: done表示本次请求结束。前端按事件类型分别渲染体验会好很多。使用请求 ID 追踪全链路。 每个请求生成一个 UUID在日志、指标采集、异常上报里带上它。流式场景中排查问题会更困难请求 ID 是唯一的线索。超时管理。 大模型接口可能长时间不返回。OpenAI Python SDK 有默认超时但流式接口需要单独设置timeout和max_retries。长任务建议使用任务队列异步处理而不是让一个 HTTP 连接等十分钟。9. 结语从 RAG 到 Agent 再到 MCP这三块内容构成了 2026 年大模型应用开发的主线能力。文章里的所有代码都可以直接复制修改使用建议你按以下顺序动手实验第一次运行搭建 FastAPI 服务调用普通非流式接口。第二次运行导入几篇真实文档测试 RAG 检索效果。第三次运行让 Agent 通过 MCP 调用工具观察它如何规划步骤。第四次运行接上流式输出在前端页面观察逐字渲染效果。整个链路跑通之后再回到业务中思考你的场景最需要的是知识问答RAG还是自动化任务Agent还是统一工具接入MCP不要一开始就把所有功能都做进去。一个好的大模型应用往往是从一个最小闭环开始逐步扩展能力边界。如果这篇文章对你有用欢迎收藏备用。下一篇我会继续拆解 RAG 评估指标、重排序算法选型以及 Agent 记忆组件的设计。有任何问题可以在评论区留言我会挑有代表性的问题展开讲。