AI Agent开发实战:LangChain、LangGraph与MCP核心要点解析
2026/9/1 12:34:56 网站建设 项目流程

2026 年再聊 AI Agent 开发,你会发现一个很有意思的现象:市面上的教程标题越来越夸张,“吊打付费”“全套实战”“从入门到企业级”,但真正把原理讲透、把代码跑通、把坑说清楚的内容反而更难找了。

这篇教程不想用“吊打”这种词来吸引眼球,只想踏实做一件事:把 LangChain、LangGraph、MCP、Agent 这四个高频关键词放在同一张技术地图里,讲清它们分别解决什么问题、如何协作、以及怎么在一个企业级最小项目中真正跑起来。

如果你正准备入门 AI Agent 开发,或者已经写了一段时间 LangChain,但总感觉停留在“调 API + 拼 Prompt”的阶段,那么这篇文章值得看完。它会帮你绕开市面教程最常见的几个认知误区,同时给出一套可以直接复用的代码骨架。

先给出全文的核心判断:LangChain 是组件库,LangGraph 是编排引擎,MCP 是工具接入协议,Agent 是最终产品形态。它们不在同一层,根本不能互相替代。很多人学了很久,其实是在把这四层混在一起学,所以才越学越乱。

1. 为什么 2026 年做 AI 应用,绕不开 Agent 架构

先回到一个根本问题:大模型的能力已经很强了,为什么还要搞 Agent?

如果你只是做一个“问答机器人”,那确实只需要一个 Prompt + 一个模型 API 就够了。但企业级 AI 应用很少停留在问答层面。真实业务往往长这样:

  • 用户问:帮我查一下华东区最近一周的订单异常情况,并给业务部门发一封摘要邮件。
  • 用户问:根据这份合同,把风险条款提取出来,写入 CRM 系统的风险台账。
  • 用户问:今天库存低于安全线的 SKU 有哪些?自动发起补货审批流。

这些请求有一个共同点:模型只靠自身参数无法完成,必须调用外部系统、读取实时数据、执行连续操作,并且在整个过程中保持状态一致。

这就是 Agent 架构存在的意义。

RAG 解决的是“模型不知道”的问题,Agent 解决的是“模型不能做”的问题。RAG 把检索结果塞进上下文,让模型回答得更准确;Agent 则把模型从“只输出文本”变成“能调用工具、能做决策、能执行多步动作”的智能体。

但“能做”和“能稳定地在生产环境做”是两回事。

从 2024 年到 2026 年,AI 应用开发的一个明显趋势是:单模型能力竞赛逐渐降温,工程架构的比拼开始成为主流。同样一个 GPT/Claude/Qwen 级别的模型,有人只能写出一个聊天 Demo,有人能做出一套带权限控制、人工审批、完整日志链路的生产系统。差距不在模型,而在工程。

于是,LangChain、LangGraph、MCP 这些工具开始浮出水面。它们不是模型,而是让模型能力“安全、可控、可复用”地输出到业务系统中的工程基础设施。

2. 四个核心概念,先把它们放对位置

很多初学者最容易犯的错,是把 LangChain 和 LangGraph 当成同一类东西,又搞不清 MCP 到底和 Agent 是什么关系。这里直接用一张表给出定位。

概念所属层级核心解决的问题常见误解
LangChain组件库提供模型调用、Prompt 管理、输出解析、记忆等基础能力以为它是一个完整的 Agent 框架
LangGraph编排引擎用状态图定义 Agent 的流程、分支、循环、并行、子图以为它只是 LangChain 的一个插件
MCP协议统一大模型与外部工具、数据源之间的接入方式以为它是一项模型能力
Agent产品形态把模型 + 工具 + 流程 + 记忆组合成能独立完成任务的系统以为它只是“调用一次工具”

2.1 LangChain 和 LangGraph 的区别,用一个例子讲清楚

LangChain 更像一个“工具箱”,里面装着各种预制件:PromptTemplateLLMOutputParserMemoryDocumentLoader等等。你用这些预制件可以快速拼出一个能跑的 Demo,但一旦需求变成“如果用户输入包含 A 就走分支一,否则走分支二;分支二执行完后要回到主流程继续跑”,LangChain 的线性 Chain 就会迅速变得难以维护。

LangGraph 换了一种思路:它把整个应用描述成一张有向图

  • 状态(State):整个流程中共享的数据结构。
  • 节点(Node):一个执行单元,可以是模型调用、工具调用、普通 Python 函数。
  • 边(Edge):节点之间的连接关系。
  • 条件边(Conditional Edge):根据当前状态决定下一步走哪个节点。

这种设计让它天然适合表达复杂业务流程。你可以在图里画循环,而循环检测、状态回滚、并行分支都变成了图本身的能力,而不是靠脆弱的 Python 控制流硬撑。

最直观的区别是:LangChain 强调链式调用,LangGraph 强调状态机。如果你的流程是“A 到 B 到 C 到 D”,两者都能做,LangChain 更快;如果你的流程是“根据条件决定走 C 还是 D,D 结束后回到 A 重新检查”,LangGraph 才是更稳的选择。

2.2 MCP 到底是个什么协议

MCP 全称 Model Context Protocol,翻译过来是“模型上下文协议”。它的作用,很像给 AI 应用装了一个通用的“USB-C 接口”。

在 MCP 出现之前,每个 Agent 要接外部工具,都得单独写一套调用逻辑:接数据库写一套数据库客户端,接飞书写一套飞书 API,接内部系统再写一套 HTTP 封装。工具越多,代码越碎,Agent 的可迁移性越差。

MCP 做的事情是:定义了一套统一的工具发现、调用、返回格式。MCP Server 负责把工具能力暴露出来,MCP Client 负责让 Agent 以标准方式去调用工具。这样,同一个 Agent 可以接入不同的 MCP Server,而不同 Agent 也可以复用同一个 MCP Server。

顺便说一句,网上常有人问“Agent Skill 和 MCP 有什么区别”。简单理解:Skill 是 Agent 的能力单元,描述了 Agent 在某个细分任务上的行为模板;MCP 是工具访问协议,解决的是 Agent 如何与外部系统通信。一个偏行为层,一个偏通信层。两者可以配合使用,但定位完全不同。

3. 2026 年做 Agent 开发,应该先用哪套技术栈

这可能是读者最关心的选型问题。我的判断非常明确:

  1. 如果你要从零搭一个企业级 Agent 应用,优先上 LangGraph。
  2. LangChain 不需要系统学,把它当“辅助工具库”即可。
  3. MCP 是必学项,因为它是 Agent 工具生态的通用语言。

为什么这么说?

因为 2026 年的 Agent 开发,核心早就不是“谁能写出一个能调工具的 Demo”,而是“谁能把它做成一个可控、可测、可运维的系统”。LangGraph 有明确的状态管理、支持断点续跑、支持人在回路(Human-in-the-loop)、支持子图复用,这些特性都是 LangChain 线性 Chain 难以直接提供的。

至于 LangChain,它攒下的组件经验仍然有价值,比如 Prompt 模板、输出解析器、文档加载器。你完全可以在 LangGraph 的某个节点内部继续使用 LangChain 组件,二者并不排斥。

选型建议可以总结成一句话:如果目标是快速验证一个点子,用 LangChain 拼;如果目标是做生产系统,用 LangGraph 编排,用 LangChain 组件辅助,用 MCP 接工具。

4. 环境准备与基础安装

本文示例假设你在一个干净的 Python 3.10 或更高版本环境里操作。具体版本以实际项目为准,但建议使用虚拟环境,避免污染全局 Python。

python -m venv .venv source .venv/bin/activate # Windows 用户请使用 .venv\Scripts\activate

接下来安装核心依赖:

pip install langchain langgraph langchain-openai mcp

补充说明:如果你用的是 OpenAI 兼容接口的模型(目前国内主流模型和本地部署模型大多支持),langchain-openai这个包可以通过设置base_url来连接不同服务,不需要改代码逻辑。

配置环境变量。新建.env文件:

OPENAI_API_KEY=your-api-key OPENAI_API_BASE_URL=https://your-model-endpoint

如果你在局域网内部署模型,还要注意两个细节:

  • 模型响应延迟通常远高于 OpenAI 官方接口,LangGraph 中要合理设置超时时间。
  • 并发能力有限时,不要把 Agent 的并行分支开得太大,否则容易把模型服务打满。

下面验证安装是否成功:

from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="your-model-name", api_key="your-api-key", base_url="https://your-model-endpoint" ) result = llm.invoke("你好,请回复:模型连接成功") print(result.content)

这一段跑通后,再往下学 LangGraph 编排。

5. LangGraph 核心机制拆解:状态、节点、边

LangGraph 和普通 LangChain 最大的不同,在于你的大脑要从“线性思维”切换到“图思维”。

5.1 State:一切共享数据的容器

在 LangGraph 中,State 是一个 TypedDict 或者 Pydantic BaseModel,是整个流程中所有节点共享的数据结构。每个节点执行完,返回的字段会更新到 State 里。

from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] next_action: str order_count: int

这里用到了Annotated[list, add_messages],它告诉 LangGraph:当多个节点往messages字段追加内容时,使用列表拼接而不是覆盖。

5.2 Node 和 Edge:把流程画出来

节点可以是一个简单的 Python 函数。函数的输入是当前 State,输出是更新后的 State 片段。

用代码演示一个最简单的“检测 -> 分析 -> 总结”流程:

from langgraph.graph import StateGraph, START, END def detect_node(state: AgentState): # 模拟检测逻辑 return {"next_action": "normal"} def analysis_node(state: AgentState): # 模拟分析逻辑 return {"order_count": 100} def summary_node(state: AgentState): # 模拟总结逻辑 return {"messages": [{"role": "assistant", "content": f"处理完成,订单数 {state['order_count']}"}]} graph = StateGraph(AgentState) graph.add_node("detect", detect_node) graph.add_node("analysis", analysis_node) graph.add_node("summary", summary_node) graph.add_edge(START, "detect") graph.add_edge("detect", "analysis") graph.add_edge("analysis", "summary") graph.add_edge("summary", END) app = graph.compile()

可以看到,LangGraph 的写法就是把 Node 和 Edge 搭成一张图,最后compile()编译成可执行应用。这张图既能被测试,也能被可视化(不过本文不展开可视化)。

5.3 条件路由:让 Agent 自己决定下一步

真实业务里几乎不存在一条直线走到底的流程。条件路由(Conditional Edge)是 LangGraph 必学点。

from langgraph.graph import StateGraph, START, END def check_node(state): # 模拟判断 order_count = state.get("order_count", 0) return {"order_count": order_count} def high_volume_node(state): return {"messages": [{"role": "assistant", "content": "订单量高,需要重点关注"}]} def normal_volume_node(state): return {"messages": [{"role": "assistant", "content": "订单量正常,无需处理"}]} def route_by_order_count(state): if state.get("order_count", 0) > 1000: return "high_volume" return "normal_volume" graph = StateGraph(AgentState) graph.add_node("check", check_node) graph.add_node("high_volume", high_volume_node) graph.add_node("normal_volume", normal_volume_node) graph.add_edge(START, "check") graph.add_conditional_edges( "check", route_by_order_count, { "high_volume": "high_volume", "normal_volume": "normal_volume" } ) graph.add_edge("high_volume", END) graph.add_edge("normal_volume", END)

这里的关键是:route_by_order_count返回一个字符串 key,mapping字典再根据这个 key 找到下一个节点。如果你发现 Agent 总是走到错误的节点,优先检查这个函数里的判断条件。

5.4 循环和循环检测:图结构天然支持,但要防死循环

LangGraph 支持循环节点。最简单的循环是:一个节点处理完后,通过条件边回到处理前的节点,反复迭代。这在 Agent 多轮反思、自我修正的场景中很有用。

但循环也带来风险:如果终止条件写错,Agent 可能无限执行下去。因此建议在每个执行单元里设置最大迭代次数。

class AgentState(TypedDict): messages: Annotated[list, add_messages] iteration: int max_iterations: int = 5 def reflect_node(state: AgentState): iteration = state.get("iteration", 0) + 1 if iteration >= state["max_iterations"]: return {"messages": [{"role": "assistant", "content": "已到达最大迭代次数,停止"}], "iteration": iteration} # 模拟反思处理 return {"iteration": iteration} def route_after_reflect(state): if state["iteration"] >= state["max_iterations"]: return "end" return "continue"

这类显式的终止条件,是生产环境 Agent 的保命配置。宁可少跑一轮,也不能让它无限循环。

6. MCP 接入:让 Agent 获得实时数据与工具能力

前面我们说的 Agent 还只能调本地函数。真正的企业级场景,Agent 必须能连接数据库、内部 API、消息系统等外部工具。这就是 MCP 的舞台。

6.1 MCP Server 最小示例

先用 FastMCP 写一个最简单的 MCP Server。假设我们提供一个订单查询工具:

from mcp.server.fastmcp import FastMCP mcp = FastMCP("order-service") @mcp.tool() def get_order_count(region: str) -> str: """查询指定区域的订单数量(示例为模拟数据)""" # 真实项目里,这里会去查数据库或调用订单中心 API mock_data = { "华东": 1580, "华南": 860, "华北": 920 } count = mock_data.get(region, 0) return f"{region} 区域当前订单量:{count}" if __name__ == "__main__": mcp.run()

这段代码的核心是@mcp.tool()装饰器。它把一个 Python 函数暴露成一个符合 MCP 协议的工具,函数签名和 docstring 会被 MCP 自动整理成工具描述,供大模型识别。

6.2 LangGraph 中调用 MCP 工具的接线方式

在 LangGraph 的某个 Node 里,我们需要启动一个 MCP Client,把 MCP 工具拉回来,再转换为 LangChain 可用的 Tool 格式。这里给出一个通用的接线示意:

from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools # 这里的 server.py 就是上面 MCP Server 的文件 server_params = StdioServerParameters( command="python", args=["mcp_server_order.py"] ) async def call_order_tool(query: str): async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools = await load_mcp_tools(session) # 将 tools 交给 Agent 节点使用 return tools

注意:MCP 的 stdio 通信方式启动了一个子进程,生产环境里要充分评估它带来的部署复杂度。如果要接入远程 MCP Server,官方也提供 Streamable HTTP 等传输方式,但配置会更复杂,建议先跑通本地再上远程。

6.3 为什么 MCP 能让“工具生态”活起来

传统开发中,每个 Agent 都要为每一个工具写一套调用代码,而 MCP 把工具调用标准化了。这意味着:

  • 同一个 MCP Server 可以被不同 Agent 复用。
  • 同一个 Agent 可以按需动态发现并挂载新的 MCP Server。
  • 前端设计工具可以暴露成 MCP,数据库可以暴露成 MCP,内部系统可以暴露成 MCP,Agent 的应用边界一下子打开了。

这也是为什么现在很多工具链都在做 MCP 支持。它本质上是在重新定义 Agent 和工具之间的连接标准。

7. 完整示例:一个带工具调用的企业级 Agent

前面把概念拆开了,现在把它们组装起来。这个示例的目标是一个企业级工单查询助手,它接收用户输入,判断是否需要查询订单数据(工具),然后根据查询结果生成回复。整个流程用 LangGraph 编排,MCP 负责提供订单工具,同时加入记忆机制。

7.1 定义 State

from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] need_tool: bool final_answer: str

7.2 创建带记忆的 LangGraph Agent

这里使用MemorySaver保存会话状态,让 Agent 能记住历史消息。在生产环境中,你可以将 MemorySaver 替换为 Redis 或数据库存储。

from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langgraph.prebuilt import ToolNode # 模拟一个订单查询工具(真实场景中可以从 MCP Server 获取) @tool def query_order_count(region: str) -> str: """根据区域查询订单数量,支持华东、华南、华北。""" data = {"华东": 1580, "华南": 860, "华北": 920} count = data.get(region, 0) return f"{region} 区域当前订单量为 {count}" tools = [query_order_count] llm = ChatOpenAI( model="your-model-name", api_key="your-api-key", base_url="https://your-model-endpoint" ) llm_with_tools = llm.bind_tools(tools) def should_use_tool(state: AgentState): """判断是否需要调用工具""" last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return "respond" def agent_node(state: AgentState): """Agent 决策节点:决定是调用工具还是直接回答""" result = llm_with_tools.invoke(state["messages"]) return {"messages": [result]} def respond_node(state: AgentState): """根据工具结果生成最终回复""" last_message = state["messages"][-1] return {"final_answer": last_message.content} graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", ToolNode(tools)) graph.add_node("respond", respond_node) graph.add_edge(START, "agent") graph.add_conditional_edges( "agent", should_use_tool, { "tools": "tools", "respond": "respond" } ) graph.add_edge("tools", "agent") graph.add_edge("respond", END) checkpointer = MemorySaver() app = graph.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "customer-service-001"}}

这段代码是整个教程最核心的部分,也是 LangGraph + Tool 调用 + 记忆的标准骨架。当你跑通它之后,后续加 MCP 工具、加多节点、加人工审批,都只是在这个骨架上做扩展。

user_input = "帮我查一下华东区的订单量" result = app.invoke( {"messages": [{"role": "user", "content": user_input}]}, config=config ) print(result["final_answer"])

第二轮提问时,因为同一个thread_id下的记忆还在,Agent 会记得刚才查过华东区订单量:

user_input2 = "那华南呢?" result2 = app.invoke( {"messages": [{"role": "user", "content": user_input2}]}, config=config )

这就是记忆的价值:Agent 不需要每次都得用户把上下文重新说一遍。

8. 运行结果与效果验证

运行上面代码,你应该能看到类似下面的输出:

Agent 第一次决策:调用工具 工具返回:华东 区域当前订单量为 1580 Agent 最终回复:华东区域当前订单量为 1580 单。

判断是否成功,可以按这个顺序检查:

  1. 模型有没有发起工具调用:如果 Agent 忽略了工具,直接回答“我无法查询”,问题大概率出在模型对工具描述的理解上,可以优化@tool里的 docstring。
  2. ToolNode 有没有真正执行:在工具函数里加一行print("tool executed:", region),确认工具节点被触发。
  3. 最终回复是否正确引用工具结果:如果 Agent 给出了和工具结果不一致的数字,检查 Prompt 或模型输出解析逻辑。

如果第一轮就失败,优先看 LLM 的调用日志和 LangGraph 的节点执行顺序,不要急着改代码。先定位是“模型没调工具”还是“工具执行抛异常”。

9. 常见问题与排查思路

问题现象可能原因排查方式解决方案
pip 安装 langgraph 时报依赖冲突langchain 和 langgraph 版本不兼容pip list查看已安装版本,pip check检查依赖统一升级到最新稳定版本,避免混装老版本
Agent 启动后不调用任何工具Prompt 中没有明确工具使用规则,或模型版本不支持 function calling打印模型返回的完整消息对象,确认tool_calls字段是否为空优化工具描述,或更换支持工具调用的模型
Agent 调用工具超时,报the agent execution provider did not respond in time模型服务响应太慢,或 MCP Server 启动时间过长检查模型服务和 MCP Server 的日志,测量接口响应耗时增加超时时间,为 MCP Server 增加预热机制
MCP Server 连接失败stdio 子进程启动失败,或参数配置错误先单独运行python mcp_server_order.py看是否报错检查 Python 环境和依赖,确认 MCP 服务可独立启动
LangGraph 节点反复执行,陷入死循环条件路由的终止条件写错,或循环节点缺少迭代上限在路由函数里打印当前 State 的迭代字段显式设置max_iterations,并在达到上限后强制END
多轮对话中记忆丢失未配置 checkpointer,或thread_id不固定检查compile()是否传入 checkpointer使用MemorySaver或替换为 Redis/数据库存储,并固定thread_id
工具返回的字段无法被模型正确解析工具返回结构复杂,超出了模型的解析能力打印模型最终回复,对比工具返回原文让工具返回结构化 JSON,并在 Prompt 中说明字段含义

10. 最佳实践与生产建议

10.1 先画图,再写代码

LangGraph 最大价值的背后,是它逼着你先把流程想清楚。强烈建议在写代码之前,用一张白纸画出:

  • 有哪些节点。
  • 每个节点需要读哪些 State 字段、输出哪些字段。
  • 哪些地方有分支,哪些地方有循环。
  • 哪些步骤允许 Agent 自主决策,哪些步骤必须人工确认。

画不清楚的流程,写出来的代码一定乱。

10.2 工具返回必须结构化

企业级 Agent 的工具返回,不要用大段文本,尽量使用 JSON。结构化数据便于模型解析,也便于输出校验和日志审计。

{ "region": "华东", "order_count": 1580, "alert_level": "high", "unit": "单", "timestamp": "2026-05-20T10:30:00Z" }

10.3 大模型擅长判断,但不擅长精确计算

不要让模型自己做订单量加总、金额计算。这类操作交给 Python 函数或 SQL 完成,模型只负责解析用户意图和生成最终文案。这样既提高准确率,也降低模型幻觉风险。

10.4 安全边界与最小权限原则

这一点值得单独强调。Agent 一旦能调用工具,就意味着它拥有了访问外部系统的能力。必须遵循以下原则:

  • 每个工具只授予完成该任务所需的最小权限,不要使用管理员账号。
  • 涉及删除、修改、审批、转账等高风险操作时,保留人工智能身份验证和人工审批环节。
  • 工具的 API Key 和敏感配置放在密钥管理系统中,不能硬编码在代码里。
  • 生产环境开启完整日志链路,记录每一次工具调用的入参、出参和耗时。

10.5 引入可观测性体系

LangGraph 应用可以接入 LangSmith 或自行打点,记录每个节点的输入输出、执行耗时、模型 token 消耗。没有可观测性的 Agent 系统,在生产环境等于盲飞。如果团队还没有这套能力,先从简单的结构化日志开始,至少保证每个节点执行时有明确的 traceId。

10.6 提供合理的并发与超时配置

compile()之后的调用里,要根据模型能力来设计并发分支数量。本地模型并发能力弱,就应串行执行或减小并行度;外部 API 响应不稳定,就应设置合理的超时和重试策略。这里最稳妥的路线是:先小流量压测,再逐步放量。

11. 总结与后续学习方向

这篇文章真正想讲清楚的,不是某一行代码的写法,而是 Agent 开发中的一个底层坐标系:LangChain 提供组件,LangGraph 提供编排,MCP 提供工具协议,Agent 是它们组合出来的产品形态。学任何一门教程时,先问自己:当前内容属于哪一层?我缺的是组件能力、编排能力,还是工具接入能力?把这个问题想清楚,学习效率会高出很多。

如果要说学习路径,我的建议是:

  1. 先用 LangGraph 手写一个最简单的带工具调用的 Agent,把 State、Node、Edge、条件路由全部跑通。
  2. 再用 MCP 把第二个外部工具接进来,体会“工具标准化”带来的开发效率提升。
  3. 接着给 Agent 加记忆和人工审批节点,理解企业级应用为什么需要 checkpointer 和 Human-in-the-loop。
  4. 最后把应用接到真实数据库和内部 API 上,引入可观测性和权限控制体系。

不用急着追每一个新概念。Agent 应用的核心复杂度在工程侧,不在模型侧。你能把“可控编排 + 工具协议 + 可观测”这条线打通,就已经超越了大多数停留在 Demo 阶段的团队。这也是“从入门到实战”真正要跨过的门槛。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询