在大模型应用开发里，LangGraph、MCP、RAG、Agent 这组概念出现的频率越来越高，但初学者最容易卡在同一个问题上：这四个词到底谁是谁，为什么要放在一起学。LangGraph 是编排框架，负责把多步骤流程组织成图；RAG 是知识增强方案，负责给模型提供外部资料；MCP 是工具接入协议，负责把外部系统标准化接进来；Agent 是基于这些组件组装出来的应用形态。它们不是竞争关系，而是同一条链路上的不同层次。

这篇博客沿着一条实操主线展开：用 LangGraph 搭一个带本地知识库检索、能通过 MCP 调用外部工具的 Agent，再从单智能体升级到多智能体协作。每一段都会给出最小可运行代码、参数解释、验证方式和排错路径。先把结论放前面：跑通 demo 只需要半天，但真正决定开发水平的是看到报错后能不能按链路把根因找出来。“3 天学会”这类说法更多是学习节奏的包装，工程上手速度取决于你对每个环节失败机制的理解程度。

1. 先理解 LangGraph、MCP、RAG、Agent 各自解决什么问题

1.1 LangGraph 是编排框架，不是对话框架

LangGraph 是由 LangChain 生态发展出来的图状态机框架。很多人一上来就把它和 LangChain 比较，其实两者的层次不一样。LangChain 的 Chain 适合描述“输入经过若干步骤得到输出”的线性流程，而 LangGraph 的核心是 StateGraph，也就是状态图。

在一个 StateGraph 里，应用逻辑被拆成节点（Node）和边（Edge）。节点是普通函数，负责一段具体逻辑；边决定节点之间的流转；条件边（Conditional Edge）根据当前状态决定下一步走哪个分支。这种结构特别适合大模型应用里常见的循环、判断、回退、人工确认、多工具调用等场景。用普通代码写这些逻辑，一旦分支变多就会越来越难维护，而图结构把流程显式画出来，调试时能清楚地看到状态在每个节点之间如何变化。

LangGraph 和 LangChain 的另一个关键区别是状态管理。LangGraph 每个节点都会接收一个状态对象，处理后再返回新的状态，状态可以在节点之间传递和累积。配合 Checkpointer，还可以把每个线程的对话状态保存下来，实现多轮对话记忆。这个区别决定了它更适合写 Agent，而不是简单的问题问答链。

1.2 RAG 解决的是模型“无资料可用”的问题

RAG（Retrieval-Augmented Generation）全称是检索增强生成。大模型的知识来自训练数据，训练数据之外的内容它并不知道，强行回答就会出现幻觉。RAG 的做法是：先把内部文档切分成小块，向量化后存入向量数据库；用户提问时，先从向量库中检索出与问题最相关的片段；再把片段拼进提示词，让模型基于这些资料回答。

这个流程的本质是“开卷考试”。模型不依赖记忆中的模糊印象，而是直接依据检索到的资料作答。RAG 解决的不只是知识时效问题，还包括企业私有知识、团队内部文档、产品手册这类模型无论如何都不可能提前掌握的內容。

1.3 MCP 解决的是工具接入标准化的问题

MCP（Model Context Protocol）是一套开放协议，它把“模型如何调用外部工具”这件事标准化了。模型要访问数据库、天气服务、内部系统，传统做法是为每个系统各写一套接入逻辑：不同的鉴权方式、不同的参数格式、不同的错误处理。接入的工具越多，重复代码越多，维护成本越高。

MCP 把工具、资源、提示词统一暴露成标准接口。客户端只需要实现一套连接逻辑，就能从不同的 MCP Server 上获取工具列表并执行工具调用。可以把它理解为工具接入层的 USB 接口，统一了接口规范，各种设备才能即插即用。

1.4 Agent 是组装出来的应用形态

Agent 不是一个独立技术，而是“大模型 + 工具 + 循环决策”的组合。模型在每一步决定要不要调用工具、调用哪个工具、如何根据工具返回结果继续行动。多智能体（Multi-Agent）则是把这个决策过程拆给多个角色，让不同的 Agent 分别负责检索、写作、查询数据库等不同职责。

1.5 四个组件的关系与学习顺序

组件 类型 解决的核心问题 通俗类比
LangGraph 流程编排框架 多步骤流程如何组织、状态如何流转 项目管理流程
RAG 知识增强方案 模型如何引用外部知识 开卷考试
MCP 工具接入协议 外部工具如何标准化接入 USB 接口规范
Agent 应用形态 模型如何自主决策并调用工具 按流程办事的执行者

推荐的学习顺序是：先用 LangGraph 写一个单 Agent，再给 Agent 增加 RAG 检索能力，再通过 MCP 接入外部工具，最后才拆分多智能体。跳过前面几步直接上多智能体，往往连状态传递和工具调用都没弄明白，排错会非常困难。

2. 环境准备：虚拟环境、依赖版本与本地大模型

2.1 环境清单

学习阶段建议全部在本地完成，不要一开始就依赖云端服务，这样能更好地区分“框架问题”和“模型问题”。

项目 推荐配置 说明
Python 3.10 或 3.11 LangGraph 依赖较多，Python 3.8 以下不建议使用
包管理 venv + pip 避免污染全局环境
大模型 Ollama 拉取 qwen2.5 系列 本地运行，支持 tool calling
Embedding 模型 nomic-embed-text 或 bge-m3 中文场景推荐 bge-m3
向量库 Chroma 单机文件型，入门最省事
调试工具 LangGraph Studio（可选） 可视化查看图执行过程

版本问题在这里要特别强调：LangGraph、LangChain、langchain-ollama、langchain-mcp-adapters 这几个包的发布节奏很快，不同版本之间的 API 可能存在差异。写作时的常用组合不代表未来仍然兼容，落地前一定要用 pip index 或官方文档确认当前稳定版本。

2.2 创建虚拟环境并安装依赖

python -m venv .venv
# Windows
.venv\Scripts\activate
# macOS / Linux
source .venv/bin/activate

python -m pip install --upgrade pip
pip install langgraph langchain-core langchain-openai langchain-ollama langchain-community langchain-text-splitters langchain-mcp-adapters chromadb

安装完成后立即固定版本：

pip freeze > requirements.txt

这一步不是形式。LangGraph 相关的报错里，很大一部分是版本混装导致的，比如 ImportError: cannot import name ... ，往往是 langchain-core 和 langgraph 的版本不匹配。把版本固定下来，后续复现问题才可能。

2.3 用 Ollama 准备本地大模型和 Embedding 模型

ollama pull qwen2.5:7b
ollama pull nomic-embed-text
ollama list

选择 7B 级别模型的原因是：本地显存占用可控，同时支持 OpenAI 风格的 tool calling 格式，适合学习 Agent 工具调用。如果机器配置较高，也可以选择更大的模型，但要注意模型能力和推理速度之间的取舍。

要注意，本地小模型和云端大模型的差距是真实存在的。同一个 LangGraph 流程，换一个模型后工具调用行为可能完全不同。排查问题时如果发现模型经常不输出 tool_calls，先不要怀疑 LangGraph，先用最小调用验证模型本身的行为。

2.4 用最小调用验证模型可用

from langchain_ollama import ChatOllama

llm = ChatOllama(model="qwen2.5:7b")
resp = llm.invoke("用一句话介绍 LangGraph")
print(resp.content)

这个简单调用能确认三件事：Ollama 服务是否在运行、模型是否已经拉取、langchain-ollama 是否正常工作。如果这一步都失败，后面的图、Agent、RAG 都无从谈起。

3. 用 LangGraph 写出第一个可运行的 Agent

3.1 State、Node、Edge 三个核心概念

LangGraph 的 State 是一个 TypedDict，描述整个流程中需要传递和累积的数据。Node 是普通函数，接收当前 State，返回更新后的 State。Edge 表示固定的流转关系，Conditional Edge 表示按条件选择下一步。

为什么要用 State 而不是直接传参？因为图里的节点可能是并列、循环、回退的，直接用函数参数传递很难表达这种灵活性。State 相当于一个全局工作台，每个节点从上面取自己需要的数据，再把自己产生的新数据放回去。

3.2 一个最小图：条件路由

下面这个例子演示了最核心的用法：模型先判断问题是否需要检索知识库，然后走不同分支。

from typing import TypedDict
from langgraph.graph import StateGraph, END

class State(TypedDict):
    question: str
    answer: str
    need_rag: bool

def decide_node(state: State):
    # 实际项目中可以用模型判断，这里先用规则演示
    state["need_rag"] = "产品" in state["question"]
    return state

def rag_node(state: State):
    # 模拟检索后的回答
    state["answer"] = "基于知识库回答：" + state["question"]
    return state

def direct_node(state: State):
    state["answer"] = "直接回答：" + state["question"]
    return state

graph = StateGraph(State)
graph.add_node("decide", decide_node)
graph.add_node("rag", rag_node)
graph.add_node("direct", direct_node)

graph.set_entry_point("decide")
graph.add_conditional_edges(
    "decide",
    lambda state: "rag" if state["need_rag"] else "direct"
)
graph.add_edge("rag", END)
graph.add_edge("direct", END)

app = graph.compile()
result = app.invoke({"question": "这个产品支持哪些功能"})
print(result["answer"])

关键点在于 add_conditional_edges 的 lambda：它根据当前 State 里的 need_rag 返回目标节点名。这就是后面 Agentic RAG 的基础，也是 LangGraph 和线性 Chain 的核心差异。

3.3 用预置 ReAct Agent 快速接入工具

如果每个工具调用都手动维护循环，代码会非常多。LangGraph 提供了 create_react_agent ，直接实现 ReAct 模式：模型思考、调用工具、观察结果、再思考。

from langchain_core.tools import tool
from langchain_ollama import ChatOllama
from langgraph.prebuilt import create_react_agent

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气信息"""
    return f"{city} 今天晴，气温 22 度"

llm = ChatOllama(model="qwen2.5:7b", temperature=0)
agent = create_react_agent(llm, [get_weather], name="weather_agent")

result = agent.invoke({"messages": [("user", "北京天气怎么样")]})
print(result["messages"][-1].content)

注意：工具函数的名字和 docstring 会被转换成大模型看到的工具 schema。函数名尽量动词开头，docstring 要写清楚“这个工具做什么、参数含义是什么”。模型判断是否调用工具，依据的就是这段描述。

3.4 用 Checkpointer 保存多轮会话状态

上面的 Agent 每调用一次 invoke 都是独立的，模型不会记得上一轮对话。要保存会话记忆，需要在编译时传入 Checkpointer，并在调用时指定 thread_id 。

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
agent = create_react_agent(
    llm,
    [get_weather],
    checkpointer=checkpointer,
    name="weather_agent",
)

config = {"configurable": {"thread_id": "user-001"}}
agent.invoke(
    {"messages": [("user", "帮我记下：产品上线时间是下周三")]},
    config=config,
)
agent.invoke(
    {"messages": [("user", "我上次说的上线时间是什么")]},
    config=config,
)

thread_id 是会话分组的键。同一个 thread_id 的多次调用共享历史消息，不同的 thread_id 互不干扰。这里用的是 InMemorySaver ，数据只存在进程内存里，服务重启就丢失。生产环境需要换成基于 PostgreSQL 或 Redis 的持久化 Checkpointer。

4. 把 RAG 检索节点接入 LangGraph

4.1 RAG 的最小链路

完整的 RAG 链路是：加载文档、切分文档、向量化、存储到向量库、检索、生成回答。前面的步骤属于离线建库，后面的检索和生成属于在线推理。很多初学者把这两个阶段混在一起，导致每次启动都重新建库，既慢又浪费资源。

4.2 文档加载、切分与向量化

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from langchain_ollama import OllamaEmbeddings

loader = TextLoader("knowledge_base.txt", encoding="utf-8")
docs = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""],
)
chunks = splitter.split_documents(docs)

embeddings = OllamaEmbeddings(model="nomic-embed-text")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

中文场景下， separators 里必须包含中文标点，否则切分器会把一句话拦腰截断。切分参数直接影响检索质量，建议先记住默认值，再用真实文档验证效果。

参数 常见值 影响
chunk_size 300 到 800 字符 太小信息不完整，太大多噪声干扰
chunk_overlap 50 到 100 字符 避免语义被切断
k 3 到 5 检索返回片段数量
score_threshold 0.3 到 0.5 过滤无关片段，需结合向量距离类型调整

4.3 在 LangGraph 里加检索节点和生成节点

把上面的 retriever 封装成图节点，再接上生成节点：

from typing import TypedDict
from langgraph.graph import StateGraph, END

class RagState(TypedDict):
    question: str
    context: str
    answer: str

def retrieve_node(state: RagState):
    docs = retriever.invoke(state["question"])
    state["context"] = "\n\n".join([d.page_content for d in docs])
    return state

def generate_node(state: RagState):
    prompt = f"""基于以下资料回答问题。
资料：
{state["context"]}

问题：{state["question"]}
回答："""
    resp = llm.invoke(prompt)
    state["answer"] = resp.content
    return state

rag_graph = StateGraph(RagState)
rag_graph.add_node("retrieve", retrieve_node)
rag_graph.add_node("generate", generate_node)
rag_graph.set_entry_point("retrieve")
rag_graph.add_edge("retrieve", "generate")
rag_graph.add_edge("generate", END)

rag_app = rag_graph.compile()
result = rag_app.invoke({"question": "产品支持哪些导出格式"})
print(result["answer"])

生成节点里的提示词决定了模型是否忠于资料。这里只写了最简版本，实际项目还要加入“资料不足时明确说不知道”的约束，避免模型在资料为空时继续编造。

4.4 从基础 RAG 升级到 Agentic RAG

基础 RAG 是固定流程，无论问题需不需要检索，都会走一遍检索。Agentic RAG 的做法是：把检索封装成一个工具，让模型自主判断是否需要检索、需要检索几次。简单问题直接回答，复杂问题多次检索，这样既减少不必要的开销，又能处理需要多步查证的问题。

from langchain_core.tools import tool

@tool
def search_knowledge(query: str) -> str:
    """从内部知识库检索与 query 相关的资料片段"""
    docs = retriever.invoke(query)
    return "\n\n".join([d.page_content for d in docs])

agent = create_react_agent(
    llm,
    [search_knowledge],
    checkpointer=checkpointer,
    name="knowledge_agent",
)

到这里，LangGraph、RAG 和 Agent 已经打通了。普通产品知识问答场景，这套结构基本够用。

5. 通过 MCP 标准协议接入外部工具

5.1 MCP 的工作原理

MCP 采用 Host、Client、Server 三层结构。Host 是运行 Agent 的应用，Client 在 Host 内部负责和 Server 建立连接、获取工具列表、执行工具调用，Server 是真正实现业务逻辑的进程或服务。传输方式主要有两种：stdio 用于本地子进程通信，streamable HTTP 用于远程服务通信。

对 LangGraph 开发者来说，MCP 带来的收益是可以复用同一个协议去接不同的工具系统。团队里如果已经有人封装好了 MCP Server，其他人直接配置连接地址或启动命令就能使用，不需要重新写一遍调用代码。

5.2 用 langchain-mcp-adapters 注册 MCP Server

langchain-mcp-adapters 的作用是把 MCP Server 暴露的工具转换成 LangGraph Tool，再交给 create_react_agent 使用。

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_ollama import ChatOllama

llm = ChatOllama(model="qwen2.5:7b")

with MultiServerMCPClient(
    {
        "db": {
            "command": "npx",
            "args": ["-y", "示例数据库mcp-server"],
            "transport": "stdio",
        },
        "weather": {
            "url": "http://127.0.0.1:8000/mcp",
            "transport": "streamable-http",
        },
    }
) as client:
    tools = client.get_tools()
    agent = create_react_agent(llm, tools, name="mcp_agent")

    result = agent.invoke({
        "messages": [("user", "查询订单表里最近 3 条订单")]
    })
    print(result["messages"][-1].content)

上面用的是一个示例包名，实际项目中要替换成团队内部已经存在的 MCP Server 包或你自己开发的 Server。Windows 系统上配置 stdio 类型时要注意： command 必须是系统能找到的可执行文件，带空格的路径要处理引号，依赖的环境变量要提前设置好，否则会直接报 ENOENT 或启动失败。

5.3 一次 MCP 工具调用的完整链路

从用户提问到最终输出，中间经历了这些步骤：

  1. 用户问题进入 Agent。
  2. 模型根据问题决定调用哪个工具，输出 tool_calls。
  3. LangGraph 在已注册的工具列表中解析出对应的 MCP Tool。
  4. MCP Client 通过 stdio 或 HTTP 调用远端 Server。
  5. Server 执行业务逻辑并返回结果。
  6. 结果作为 ToolMessage 回传给模型。
  7. 模型基于工具结果生成最终回答。

调试时建议在每一步打印或记录消息类型。如果最终回答说“查询失败”，先看 ToolMessage 里实际返回了什么，往往错误原因离答案只差这一步日志。

5.4 MCP 配置的常见错误

  • 本地 Server 进程没有启动，调用时出现 connection refused。
  • 传输方式写错，stdio 和 HTTP 的配置字段完全不同。
  • Server 返回超时，Agent 等待时间过长。
  • 两个 Server 暴露了同名工具，导致工具解析冲突。
  • API Key 或数据库口令写在代码里，而不是通过环境变量注入。

前三种属于配置问题，后两种属于工程问题。生产环境接入 MCP Server 时，一定要把鉴权、密钥管理、超时重试全部纳入设计，不要把内部系统凭据直接暴露给 Agent 流程。

6. 多智能体编排：Supervisor 与 Worker 协作

6.1 为什么单个 Agent 不够

单个 Agent 可以挂载很多工具，但工具一多，模型在每一步选择工具时的出错率会上升，同时大量工具 schema 会占用上下文窗口，导致成本上升、响应变慢。多智能体把职责拆开：检索 Agent 只面对检索工具，写作 Agent 只面对写作任务，查询 Agent 只面对数据库工具。每个智能体要处理的上下文更小，决策更稳定。

6.2 常见协作模式对比

模式 结构特征 适用场景
Supervisor + Worker 一个协调者负责拆任务和汇总 任务可拆解、结果可合并
Pipeline 上游输出直接作为下游输入 固定顺序的处理流程
分层式 多个 Supervisor 形成层级 大规模复杂系统
对话式 多个 Agent 互相讨论 研究探索类任务，成本较高

初学者建议先从 Supervisor + Worker 开始，它最容易理解，也最容易用 LangGraph 表达。

6.3 用 LangGraph 实现 Supervisor + Worker

下面这个例子把前面写好的知识 Agent 和写作 Agent 作为两个 Worker，由一个 Supervisor 节点负责分配任务：

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import create_react_agent

class WorkState(TypedDict):
    task: str
    results: dict

knowledge_agent = create_react_agent(llm, [search_knowledge])
writer_agent = create_react_agent(llm, [])

def supervisor_node(state: WorkState):
    # 实际项目中可以在这里调用模型拆分任务
    state["results"] = {}
    return state

def knowledge_worker_node(state: WorkState):
    resp = knowledge_agent.invoke({
        "messages": [("user", f"请检索资料：{state['task']}")]
    })
    state["results"]["knowledge"] = resp["messages"][-1].content
    return state

def writer_worker_node(state: WorkState):
    material = state["results"].get("knowledge", "")
    resp = writer_agent.invoke({
        "messages": [("user", f"基于以下资料写总结：\n{material}")]
    })
    state["results"]["writer"] = resp["messages"][-1].content
    return state

def aggregate_node(state: WorkState):
    state["results"]["final"] = state["results"].get("writer", "")
    return state

graph = StateGraph(WorkState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("knowledge_worker", knowledge_worker_node)
graph.add_node("writer_worker", writer_worker_node)
graph.add_node("aggregate", aggregate_node)

graph.set_entry_point("supervisor")
graph.add_edge("supervisor", "knowledge_worker")
graph.add_edge("knowledge_worker", "writer_worker")
graph.add_edge("writer_worker", "aggregate")
graph.add_edge("aggregate", END)

multi_agent = graph.compile(checkpointer=checkpointer)
result = multi_agent.invoke({"task": "整理产品 v2.0 的发布说明"})
print(result["results"]["final"])

这里两个 Worker 是串行执行的。如果任务可以并行，可以用 add_edge 从 Supervisor 同时连到两个 Worker，再让两边汇聚到同一个聚合节点。LangGraph 支持这种扇出汇聚结构，但要注意共享 State 的写入冲突，不同 Worker 最好写不同 key。

6.4 状态共享与长期记忆

在多智能体里，State 是所有 Agent 都能看到的工作台。不要把大段文本随便塞进 State，否则每轮调用都会把无用数据带给所有 Worker。更合理的做法是只传引用、摘要和必要结果，详细内容放在检索结果或消息历史里。

会话级记忆用 Checkpointer 加 thread_id ，跨会话长期记忆则需要引入 Store。LangGraph 的 Store 可以保存跨线程的键值数据，例如用户的长期偏好、历史结论。生产环境需要把 Store 落到持久化存储上，避免重启丢失。

7. 验证与排错：从现象倒推根因

7.1 验证的标准不是“能跑”

一个 Agent 能启动、能回答，不代表流程正确。要逐段验证四件事：模型是否真的输出了预期的 tool_calls、工具是否真的被执行、工具结果是否进入了下一轮上下文、最终回答是否基于工具结果或检索资料生成。建议每一步都打印关键消息类型和内容，或者使用 LangGraph Studio 观察节点级执行情况。

7.2 按链路顺序排查

排错时不要跳跃式猜测，按下面顺序检查：

  1. 输入和提示词是否正确。
  2. 依赖版本是否匹配， requirements.txt 是否和当前环境一致。
  3. 模型本身是否能正常调用工具。
  4. 向量库检索结果是否为空或是否相关。
  5. MCP Server 是否能单独连通。
  6. 图的状态和 Checkpointer 配置是否正确。

这个顺序是从“最容易确认”到“最难确认”。版本问题查一次 pip list 就能确认，模型行为则需要构造最小用例验证。

7.3 常见报错对照表

现象 常见原因 检查方式 处理建议
ImportError: cannot import name langchain 与 langgraph 版本不匹配 pip list 锁定兼容版本
模型始终不调用工具 模型不支持 tool calling，或提示词没写清楚 打印原始 messages 换模型或改进工具描述
Could not resolve tool 工具未传入当前 Agent 检查 tools 列表 确认节点真正拿到工具
Connection refused MCP Server 未启动或端口错误 检查进程和端口 先单独启动 Server 再接入
检索结果为空 文档切分失败或 embedding 模型不匹配 打印 retriever.invoke 返回值 检查文档编码和向量库
多轮对话记不住 未设置 Checkpointer 或 thread_id 不一致 查看编译配置和调用配置 统一 thread_id

7.4 可复用的排错清单

  • [ ] pip freeze 与 requirements.txt 是否一致。
  • [ ] Ollama 本地模型是否已拉取，服务是否在运行。
  • [ ] 模型在最简调用下是否能输出 tool_calls。
  • [ ] retriever.invoke 是否返回了相关片段。
  • [ ] MCP Server 能否脱离 Agent 单独调用成功。
  • [ ] 每次调用传入的 thread_id 是否一致。
  • [ ] 最终
# MCP # RAG
Logo
码道开发者社区

码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践

更多推荐

cover

Claude API + Python实战：结构化输出后，如何拦住“看起来正确”的JSON

avatar 码道开发者社区
cover

GLM-5.3-Flash深度解析：3200亿参数只卖Opus四十分之一，国产算力首次跑通前沿模型

avatar 码道开发者社区
cover

AI Agent的应用场景

avatar 码道开发者社区