如果你正在准备大模型应用开发又绕不开LangGraph、MCP、RAG、Agent这几个热词那么这篇教程可以帮你把概念、选型、实战和避坑串起来。题目标题里的“3 天学会”不用太当真真正重要的是先跑通一条完整链路再基于这条链路去扩展自己的业务场景。这篇文章面向的目标读者是已经会用 Python 调用大模型 API知道什么是 prompt、什么是 embedding但还没系统做过 Agent 编排和知识库增强的开发者。读完你会掌握LangGraph 解决什么问题它和 LangChain 有什么区别。MCP 是什么为什么它让工具接入变得更标准化。RAG 在 LangGraph 工作流里怎么落地怎么和 Agent 结合。一个从零到一的实战项目本地启动、接口调用、批量任务编排、效果验证。常见报错、性能排查和工程化建议。1. 核心能力速览能力项说明技术栈LangGraph、LangChain、MCP、RAG、Agent 多智能体主要功能大模型工作流编排、工具调用、知识库检索增强、多智能体协作开发语言Python推荐硬件普通开发机即可纯 API 调用无需 GPU若本地跑模型需按模型大小准备显存显存需求不确定取决于使用的 LLM/Embedding 模型Qwen 7B 量化和 MiniLM embedding 在 8G 显存可跑实际需测试启动方式命令行 / Python 脚本 / LangGraph Studio / API 服务是否支持 API支持可通过 FastAPI 包装成 HTTP 接口是否支持批量任务支持LangGraph 支持图状态管理和循环可批量处理输入适合场景智能问答、RAG 知识库、Agent 工作流、多智能体协作系统这里先给结论如果你只想写一个“调用大模型 检索 回复”的 demoLangChain 足够。如果你的业务里存在条件分支、循环、人工确认、多智能体协作、需要可视化调试LangGraph 更合适。它不是替代 LangChain而是在 LangChain 之上提供“图状态编排”能力两者通常组合使用。2. LangGraph、MCP、RAG、Agent 到底是什么2.1 LangGraph把大模型流程变成一张图LangGraph 是 LangChain 社区推出的低层编排框架核心思路是把应用逻辑建模为一张有向图。图里面有节点Node和边Edge节点可以是“调用大模型”“检索知识库”“执行工具”“人类审批”等任何函数边决定下一步走到哪个节点。和普通链式调用相比LangGraph 的差异点在于支持循环和条件分支这是 Agent 自动决策的基础。有内置状态管理每个节点可以读写共享状态。支持持久化和人工介入适合长时间运行的任务。支持图的可视化和逐步调试。虽然 LangChain 也支持 Agent 和链但当流程复杂到“模型决定下一步动作再根据动作结果继续循环”时LangGraph 的图结构更清晰也更容易排查问题。2.2 MCP统一工具接入协议MCPModel Context Protocol是一个开放协议用来统一大模型应用与外部工具、数据源之间的连接方式。你可以把它理解为“AI 应用里的 USB 接口”。传统做法是每个工具写一个本地函数再手动注册给模型。MCP 的做法是工具方实现一个 MCP Server应用方通过 MCP Client 连接之后动态发现工具、调用工具。这样一套协议可以复用换数据库、换文件系统、接第三方服务逻辑都类似。在 LangGraph 里你可以把 MCP 工具包装成节点也可以直接在 Agent 节点里通过工具调用机制使用 MCP Server。MCP 的价值在工程化上更明显尤其是团队内多个项目共享工具时。2.3 RAG给模型外挂知识库RAGRetrieval-Augmented Generation是目前最实用的“私有知识库问答”方案。原理不复杂先把文档切块、向量化存入向量数据库用户提问时将问题向量化检索相似度最高的文档片段把检索结果和问题一起交给大模型生成回答。RAG 解决的核心问题是让模型知道训练数据里没有的内容同时减少幻觉。配合 Agent 后还可以让模型自己判断“是否需要检索”“检索什么”进一步提升回答质量。2.4 Agent 多智能体让多个角色协作Agent 是对大模型能力的封装不仅会“聊天”还能“决策 行动”。一个 Agent 节点通常包含模型 提示词 工具集合。多智能体是这个思路的扩展定义多个不同角色的 Agent比如“检索助手”“写作助手”“审核助手”由编排者决定任务的分配和流转。LangGraph 天然支持这种多节点协作每个 Agent 是图里的一个节点节点之间的边就是协作关系。所以这套技术栈的组合方式可以简单概括为用 LangGraph 编排流程。用 RAG 提供业务数据。用 MCP 接入外部工具。用 Agent 让模型具备决策和行动能力。3. 适用场景与使用边界3.1 适合什么场景企业内部知识库问答把产品文档、制度文件、技术资料向量化通过 RAG 实现专属问答。客服工单处理Agent 自动理解问题、检索答案、调用工单接口必要时转人工。内容生产辅助多智能体协作完成选题、素材检索、初稿生成、校对。数据分析助手通过 MCP 连接数据库Agent 根据自然语言问题生成查询并返回结果。长任务自动化用 LangGraph 的循环和状态管理实现多步任务比如“搜索 → 总结 → 生成报告”。3.2 不适合什么场景简单的单轮问答直接调用 API 更轻量没必要上 Agent 和图编排。对延迟极其敏感的高并发查询RAG Agent 的链路会明显增加耗时。无法接受模型输出随机性的业务需要先做好结果校验和人工审核机制。没有私有化需求、数据完全公开的场景直接用在线知识库产品可能更省事。3.3 版权、隐私与安全边界使用 RAG 时知识库内容来源必须合规不要上传未经授权的文档、个人隐私数据、商业机密。涉及人脸、声音、版权素材等场景时必须确认授权。Agent 调用外部接口尤其是写操作接口时要加权限控制和人工确认防止误操作。大模型输出不代表事实法律、医疗、金融等领域的生成结果必须人工复核。本地部署模型时注意模型许可证和商用条款。4. 环境准备与前置条件在开始写代码之前先做一次环境检查。4.1 基础环境清单项目要求操作系统Windows 10/11、macOS、Linux 均可Python3.9 及以上推荐 3.10 或 3.11包管理pip 或 uvAPI KeyOpenAI / Anthropic / 国内大模型服务商或者用本地部署模型向量数据库简单场景用 Chroma / FAISS生产场景用 Milvus / pgvectorGPU可选纯 API 模式不需要本地模型需要按模型选择4.2 安装依赖建议先创建虚拟环境python -m venv venv source venv/bin/activate # Linux/macOS venv\Scripts\activate # Windows安装核心依赖pip install langgraph langchain langchain-openai langchain-community pip install chromadb faiss-cpu pip install mcp fastapi uvicorn python-dotenv注意实际安装时以官方最新版本为准不同版本之间 API 可能有细微差异。如果遇到兼容性问题锁定 langchain 和 langgraph 的大版本号。4.3 模型选择如果只是学习流程优先用 API 模型例如 GPT 系列、Claude、国内厂商的大模型 API。都支持 OpenAI 兼容格式用 LangChain 的ChatOpenAI类传入 base_url 即可。如果你想完全本地部署可以尝试Qwen2.5-7B-Instruct的量化版、ChatGLM系列或者更轻量的模型。Embedding 模型推荐BAAI/bge-small-zh这类轻量模型本地跑不费力。本地模型启动方式可以是 Ollama、vLLM、Xinference 等。LangChain 有对应的接入类配置base_url就能用。5. 实战项目搭建一个“知识库 Agent”下面我们实现一个完整的实战项目一个带 RAG 检索和 MCP 工具调用的 Agent。用户提问后Agent 先判断是否需要检索知识库再决定是否调用工具最后生成回答。整个流程用 LangGraph 编排。5.1 项目结构langgraph_rag_agent/ ├── .env ├── requirements.txt ├── main.py # 入口脚本支持命令行提问 ├── graph.py # LangGraph 图定义 ├── retriever.py # RAG 检索模块 ├── mcp_tools.py # MCP 工具封装 └── api_server.py # FastAPI 接口服务5.2 配置环境变量创建.env文件# 选择 OpenAI 兼容接口 OPENAI_API_KEYyour-api-key OPENAI_API_BASEhttps://api.example.com/v1 LLM_MODELgpt-4o-mini # Embedding 模型 EMBEDDING_MODELtext-embedding-3-small如果使用本地模型OPENAI_API_BASE指向本地服务地址例如http://127.0.0.1:8000/v1。5.3 编写 RAG 检索模块retriever.py负责文档加载、切块、向量化和检索。import os from dotenv import load_dotenv from langchain_community.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma load_dotenv() def build_vectorstore(doc_pathdocs, persist_dir./chroma_db): loader TextLoader(doc_path, encodingutf-8) documents loader.load() splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50 ) docs splitter.split_documents(documents) embeddings OpenAIEmbeddings(modelos.getenv(EMBEDDING_MODEL)) vectorstore Chroma.from_documents( docs, embeddings, persist_directorypersist_dir ) return vectorstore def get_retriever(): embeddings OpenAIEmbeddings(modelos.getenv(EMBEDDING_MODEL)) vectorstore Chroma( persist_directory./chroma_db, embedding_functionembeddings ) return vectorstore.as_retriever(search_kwargs{k: 4})5.4 封装 MCP 工具mcp_tools.py里演示如何通过 MCP 协议连接一个简单的工具服务。这里用一个本地 http 服务模拟 MCP 工具调用你也可以换成真实的 MCP SDK。import requests def call_mcp_tool(tool_name: str, params: dict) - str: 实际项目中这里应该使用 mcp 官方 SDK 建立连接。 下面只是一个通用 HTTP 调用模板。 mcp_server_url http://127.0.0.1:8080/mcp payload { tool: tool_name, params: params } response requests.post(mcp_server_url, jsonpayload, timeout30) response.raise_for_status() return response.json().get(result, )在 LangGraph 中工具节点可以直接调用这个函数。5.5 定义 LangGraph 图在graph.py中定义状态、节点和条件边。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage from retriever import get_retriever from mcp_tools import call_mcp_tool class AgentState(TypedDict): question: str context: str answer: str needs_tool: bool def decide_node(state: AgentState): 让模型判断是否需要检索和调用工具 llm ChatOpenAI(modelgpt-4o-mini) messages [ SystemMessage(content判断用户问题是否需要检索知识库或调用工具回答 yes 或 no。), HumanMessage(contentstate[question]) ] decision llm.invoke(messages).content.strip().lower() return { context: , needs_tool: yes in decision } def retrieve_node(state: AgentState): retriever get_retriever() docs retriever.invoke(state[question]) context \n\n.join([doc.page_content for doc in docs]) return {context: context} def tool_node(state: AgentState): 这里根据场景调用 MCP 工具示例为查询天气 result call_mcp_tool(get_weather, {city: state[question]}) return {context: result} def answer_node(state: AgentState): llm ChatOpenAI(modelgpt-4o-mini) context state.get(context, ) prompt f基于以下信息回答用户问题。 如果信息不足请明确说明。 参考信息 {context} 用户问题{state[question]} messages [HumanMessage(contentprompt)] answer llm.invoke(messages).content return {answer: answer} def build_graph(): graph StateGraph(AgentState) graph.add_node(decide, decide_node) graph.add_node(retrieve, retrieve_node) graph.add_node(tool, tool_node) graph.add_node(answer, answer_node) graph.set_entry_point(decide) graph.add_conditional_edges( decide, lambda state: tool if state[needs_tool] else retrieve, {tool: tool, retrieve: retrieve} ) graph.add_edge(retrieve, answer) graph.add_edge(tool, answer) graph.add_edge(answer, END) return graph.compile() agent_app build_graph()上面这个图包含四个节点decide节点先让模型判断走检索还是调工具最后统一交给answer回答。实际项目中你可以根据业务增加更多分支和节点。5.6 命令行入口main.pyfrom graph import agent_app def main(): question input(请输入问题) result agent_app.invoke({question: question}) print(\n 回答 ) print(result[answer]) if __name__ __main__: main()运行python main.py第一次运行会构建向量库如果还没有索引文档需要先执行文档加载。可以在retriever.py里加一个入口函数。6. 功能测试与效果验证6.1 测试维度设计测试点输入示例预期结果判断标准基础知识问答“LangGraph 的状态是什么”模型给出可理解解释答案不偏离基本概念RAG 检索增强“公司年假制度是什么”答案引用知识库内容答案包含文档中的关键条款工具调用“北京天气怎么样”触发 MCP 工具节点日志中出现工具调用记录多轮对话连续追问状态保留上下文不丢失批量问题列表输入全部完成无异常中断6.2 验证 RAG 是否真正生效一个常见误区是模型在回答里提到“根据文档”但实际并没有检索到正确内容。验证方法在retrieve_node里打印context前 200 字。手动检查检索结果的相关性。尝试问一个知识库里有、但模型预训练数据里不太可能有的问题比如内部系统名称、具体的编号。6.3 验证条件分支是否生效在decide_node里加入日志print(f决策结果: {decision})然后分别问“什么是 RAG”和“查一下天气”观察流向是retrieve还是tool。6.4 常见失败原因问题现象可能原因排查方式解决方案问答结果全是通用知识RAG 检索未生效或相关性低打印 context 内容调大 k 值调整 chunk_size模型频繁调用工具决策提示词不明确查看决策日志改写 System Prompt限制触发条件工具调用超时MCP Server 未启动或地址错误检查服务日志确认工具服务在线报错KeyError: context状态字段未初始化检查初始化状态invoke时传入完整状态回答为空LLM 返回空字符串检查模型 API 返回增加失败重试逻辑本地模型回答质量差模型过小或 prompt 不清晰测试不同模型换更大模型或优化 prompt7. 接口 API 与批量任务7.1 用 FastAPI 包装成 HTTP 服务api_server.py可以把 LangGraph 图包装成接口方便前端和其他服务调用。from fastapi import FastAPI from pydantic import BaseModel from graph import agent_app app FastAPI() class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): answer: str context: str app.post(/api/chat, response_modelQueryResponse) async def chat(req: QueryRequest): result agent_app.invoke({question: req.question}) return QueryResponse( answerresult[answer], contextresult.get(context, ) ) if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)启动python api_server.py调用curl -X POST http://127.0.0.1:8000/api/chat \ -H Content-Type: application/json \ -d {question: 什么是RAG}Python 调用import requests url http://127.0.0.1:8000/api/chat payload {question: LangGraph 和 LangChain 有什么区别} response requests.post(url, jsonpayload, timeout60) print(response.json())7.2 批量任务编排LangGraph 支持用循环节点处理批量输入。最简单的方式是外部循环遍历问题列表逐个调用图questions [ 公司年假制度是什么, 如何申请报销, MCP 协议解决什么问题 ] results [] for q in questions: result agent_app.invoke({question: q}) results.append({question: q, answer: result[answer]}) for r in results: print(r[question], -, r[answer])如果你需要并发批量处理可以用 Python 的ThreadPoolExecutor。但要注意如果图内部有可变全局状态或者数据库连接非线程安全需要加锁或使用独立连接。from concurrent.futures import ThreadPoolExecutor, as_completed def handle(q): result agent_app.invoke({question: q}) return q, result[answer] questions [问题1, 问题2, 问题3] with ThreadPoolExecutor(max_workers4) as executor: futures [executor.submit(handle, q) for q in questions] for future in as_completed(futures): q, answer future.result() print(q, answer)7.3 失败重试建议接口层对超时和 5xx 错误做指数退避重试。图内部在节点函数内捕获异常返回错误消息而不是直接中断。批量任务记录每个任务的输入、输出、异常和耗时方便事后排查。8. 资源占用与性能观察8.1 观察什么纯 API 模式下LangGraph 本身占用很小主要看服务进程的 CPU 和内存。如果使用本地模型显存占用取决于模型大小、量化方式、推理框架和并发数。RAG 检索阶段Embedding 模型推理会占用少量 CPU/GPU 资源。向量检索的耗时取决于向量库规模和索引类型。8.2 如何测量在启动脚本里加一个装饰器统计每个节点耗时import time from functools import wraps def timer(func): wraps(func) def wrapper(*args, **kwargs): start time.time() result func(*args, **kwargs) print(f[{func.__name__}] 耗时 {time.time() - start:.2f}s) return result return wrapper然后给节点函数加上timer就能看到每个阶段的时间分布。重点观察decide_node的模型调用耗时。retrieve_node的向量检索耗时。answer_node的生成耗时。优化思路如果retrieve慢考虑减少文档数量、使用更快的向量库、预检索缓存。如果 LLM 调用慢考虑换小模型、减少输出长度、增加流式输出。如果多智能体协作慢减少不必要的中间节点或合并简单节点。8.3 降低显存占用的通用方式使用量化模型例如 4bit 量化。选择体积小的 Embedding 模型。控制并发请求数。分批处理文档不要一次性加载到内存。使用 vLLM 或 Ollama 等支持 KV Cache 优化的推理框架。9. 常见问题与排查方法整理一份 LangGraph 与 MCP 开发中常见的报错排查表。问题现象可能原因排查方式解决方案ModuleNotFoundError: langgraph未安装依赖pip list检查pip install langgraphAPI Key 不生效环境变量未加载打印os.getenv检查.env文件和load_dotenv端口被占用服务已在运行lsof -i:8000(Linux/macOS) 或netstat -ano(Windows)换端口或关闭旧进程模型回复不相关内容prompt 或检索上下文质量差查看日志中的 context优化检索分块、重写 promptMCP 工具找不到MCP Server 未注册或地址错误直接请求工具地址测试确认工具服务在线并配置正确LangGraph 图编译失败节点名或边定义错误查看异常栈检查节点名是否一致边的方向是否正确显存不足 OOM模型太大或并发太高查看 GPU 显存占用换小模型、量化、减少并发批量任务中断某个输入触发异常查看失败日志增加 try/except 并保存失败样本输出质量不稳定模型温度过高检查生成参数降低 temperature增加确定性agent terminated due to errorAgent 执行过程中模型返回异常或工具失败查看详细日志和最后的状态增加错误处理节点或重试逻辑检查工具返回格式9.1 排查思路如果组件比较多建议按层排查先测试 LLM 本身能否正常响应。再测试向量库检索直接打印检索结果。然后测试 MCP Server 的连通性。最后把所有环节串起来在 LangGraph 每个节点加日志。10. 最佳实践与使用建议10.1 工程化落地建议第一次先跑通最小链路不要一上来就设计复杂的图结构。一个检索 → 回答的图只有两个节点先验证再扩展。保留一套最小可运行配置包括依赖版本、环境变量示例、测试数据方便团队成员快速上手。模型文件、知识库文档、输出结果分开目录管理避免混在一起。批量任务必须加日志和失败重试机制记录每个任务的输入、输出、异常和耗时。接口服务要限制访问范围至少设置 API Key 和请求频率限制不要直接暴露到公网。涉及人脸、声音、版权素材、个人数据时必须确认授权遵守相关法律法规。10.2 提示词设计Agent 的决策质量很大程度取决于提示词。建议明确角色和任务范围。给出工具使用规则什么情况用、什么情况不用。要求模型在信息不足时直接说明不要编造。对敏感操作要求模型输出“需要人工确认”的标记。多轮场景下把历史对话压缩后放入状态避免上下文超长。10.3 数据与流程管理知识库文档要定期更新新增内容后重新做向量化。删除旧版本文档时记得清理向量库中的对应 embedding。对每次问答加上 trace_id方便追踪整个图的运行路径。发布到生产环境前用一组固定的评测问题做回归测试观察回答质量是否下降。11. 总结与下一步刚才这套从零开始的实战流程核心是让你先看懂 LangGraph 的图结构再把 RAG 检索和 MCP 工具调用挂到图里。这里最值得你亲手验证的不是代码能不能跑而是模型“决策”这一步是否可靠。接下来你可以依次做这样几件事替换成自己的知识库文档测试 RAG 检索质量。接入一个真实的 MCP Server例如数据库查询或文档读取工具跑通工具调用。增加一个人工审核节点让 Agent 在执行写操作前等待确认。把api_server.py部署到服务器配合前端页面做一个小产品原型。要避开的坑也很明确不要让 Agent 过度调用工具不要以为 RAG 检索出来就一定是对的不要忽略异常处理和日志采集。只要先跑通最小链路再逐步加功能LangGraph 这套技术栈完全可以支撑起企业级的 Agent 应用。如果这篇教程对你有帮助建议先收藏后面按照章节一步步跑。遇到具体报错优先检查依赖版本、环境变量、端口占用和 MCP Server 状态大部分问题都能从这个方向解决。