LangChain、LangGraph与MCP协同构建AI Agent工具调用链路
2026/9/7 13:05:59 网站建设 项目流程

先想清楚一个问题:AI大模型发展到 2026 年这个阶段,很多人已经不再讨论“大模型能做什么”,而是开始讨论“怎么让大模型稳定地完成一套真实任务”。LangChain、MCP、LangGraph 和 Agent 这四个词之所以频繁一起出现,是因为它们分别解决了 Agent 开发链路中的不同环节:LangChain 负责聚合模型和工具能力,LangGraph 负责把执行流程编排成有状态、可控制的图,MCP 则用一套标准协议让外部工具接入 Agent 环境。对于一个刚开始做 AI大模型应用开发的人来说,最容易走弯路的地方不是某个 API 不会调,而是把这三者混在一起用,导致代码越写越复杂,最后既无法调试,也不敢上生产。

这篇文章会从概念开始讲清楚三者边界,然后给出一套可以直接运行的最小案例:先用 LangGraph 搭建一个能调用内置工具的 Agent,再通过 MCP 接入外部工具,最后配合运行验证、日志检查和常见问题排查,形成一条完整的 Agent 工具调用链路。适合正在学习 LangChain 入门的开发者,也适合已经在用 LangChain 做 RAG、智能客服、自动化任务,但想把执行流程从“一次性链式调用”升级为“可控 Agent 编排”的工程师。

1. 先从 Agent 的定位开始:LangChain、LangGraph 和 MCP 分别解决什么问题

1.1 Agent 到底是什么,和普通函数调用有什么区别

一个最常见的误区是把“会调用工具的模型”直接等同于 Agent。模型本身不会思考要不要调用工具,它只会根据输入文本生成输出。Agent 的本质是:在程序里定义一个循环,让模型在每一轮判断“当前是否需要调用工具”,如果需要就生成工具调用参数,程序执行工具后把结果返回给模型,模型再根据结果决定是继续调用还是给出最终答案。

用一个实际场景说明。假设你让大模型查一下本地服务器的磁盘剩余空间,再判断是否需要清理日志。如果只是普通函数调用,你会先写死调用命令、拿到输出,再让模型整理结果。这个过程里,模型不参与决策。而 Agent 模式下,你只给模型一个“查询磁盘”工具和一个“清理日志”工具,模型的回答可能变成:先查询磁盘,发现占用率超过 90%,然后调用清理工具,最后总结清理效果。每一步是否执行、按什么顺序执行,由模型根据上下文动态决定。

这个动态决策能力强,但代价是执行路径不稳定。同一个问题连续跑两次,可能调用顺序不同,也可能多调了一个工具。正因如此,Agent 不能只靠“循环调用”实现,还需要有明确的流程、状态、终止条件和异常处理。LangGraph 解决的就是这部分问题。

1.2 LangChain 是工具集,LangGraph 是编排引擎,MCP 是工具接入协议

LangChain 出现得最早,核心价值是统一了 Prompt、模型、输出解析器、向量库、文档加载器和工具的接口。让开发者可以用差不多的写法切换不同模型、不同向量库,降低组件之间的耦合。但 LangChain 自己的 Chain 机制在执行高度动态的 Agent 流程时不够灵活,因为它更适合串行或并行的固定链路,而不是让模型自主决定下一步。

LangGraph 是专门做流程编排的。它把 Agent 执行过程建模成一张有向图:节点是“模型调用”“工具执行”“用户确认”这类操作,边是状态转移条件。每次执行后,所有关键信息都会写入一个共享状态对象,后续节点能读取并修改状态。这样 Agent 的每一步都有了可观察、可回放、可中断的基础。

MCP,全称 Model Context Protocol,解决的是“工具接入标准”问题。以前每个 Agent 框架都要自己定义工具描述、参数格式和调用方式,换一个框架就得重写适配层。MCP 把工具暴露成标准化的 server,Agent 通过 client 连接 server,按统一格式发现工具、校验参数、执行调用。简单理解:LangChain 提供零件,LangGraph 负责装配和调度,MCP 统一了零件接口。三者不是重叠关系,而是互相配合的关系。

2. 动手前先分清 LangChain 和 LangGraph 的边界,避免装错依赖

2.1 两个库的定位差异

很多初学者直接跑到 LangChain 文档里搜索 Agent 相关内容,结果看到一堆 deprecation 提示,再看 LangGraph 文档又发现命名风格很像,更难分清该用哪个。

维度LangChainLangGraph
核心定位组件集成框架有状态 Agent 流程编排框架
主要对象Prompt、Model、Retriever、ToolStateGraph、State、Node、Edge
执行方式Chain 串行或并行调用图结构循环执行,状态驱动
适合场景RAG、文档处理、固定流程多步推理、动态选工具、人工审核、多 Agent
是否适合做复杂 Agent可以,但流程控制较弱更适合,支持循环、条件分支、中断恢复

需要强调的是,两者关系不是“谁替代谁”。LangGraph 运行时本身也依赖 LangChain 的模型封装和工具定义接口。安装时通常需要同时安装langchainlangchain-openailanggraph

2.2 学习环境准备:虚拟环境和基础依赖

推荐使用 Python 3.10 以上的独立虚拟环境,避免依赖冲突。这里以常见命令为例:

python3 -m venv .venv source .venv/bin/activate pip install --upgrade pip pip install langchain langchain-openai langgraph pip install langchain-mcp-adapters mcp

如果原始教程发布时间较早,版本组合可能和当前最新版有差异。安装完成后先查看实际安装版本,再继续后面的编码:

pip show langchain langgraph langchain-openai langchain-mcp-adapters mcp

这里要注意一个坑:不要盲目安装最新主版本,而要确认langgraphlangchain-core的兼容关系。LangGraph 会依赖 langchain-core,如果之前环境里已经装过旧版 langchain,升级后可能出现函数入参签名不一致、工具装饰器失效等问题。常见做法是安装到一个全新的虚拟环境,再把依赖版本固化成requirements.txt

pip freeze > requirements.txt

注意:实际项目落地前,先打开requirements.txt确认版本号都能被 PyPI 解析。不要直接用pip install langgraph后再手动改代码去适配不兼容版本,那样排查成本很高。

2.3 一段代码看出两者的协作关系

如果你还不确定两者边界,下面这个最小示例可以直观说明。它没有写完整 Agent,只演示了“定义工具 -> 绑定模型 -> 模型决定调用工具”这一小段链路。

from langchain_openai import ChatOpenAI from langchain_core.tools import tool @tool def get_cpu_usage() -> float: """返回当前进程所在机器的 CPU 使用率,单位百分比。""" # 演示代码,实际应读取系统监控数据 return 42.5 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) tools = [get_cpu_usage] llm_with_tools = llm.bind_tools(tools) response = llm_with_tools.invoke("现在 CPU 占用情况怎么样?") print(response.tool_calls)

这个例子中,@tool是 LangChain 定义工具的方式,bind_tools把工具描述传给模型。模型如果觉得需要查询,会在返回内容里带出tool_calls,也就是“工具名 + 参数”。到这里为止,流程还是单轮的,没有循环,也没有状态。要让它真正变成 Agent,就需要 LangGraph 把“模型返回”作为图的一个节点,把“工具执行”作为另一个节点,再根据tool_calls是否为空决定进入结束还是继续循环。

3. 用 LangGraph 从零搭建一个能调用工具的最小 Agent

3.1 项目结构和文件规划

为了避免后续扩展时逻辑混乱,建议按模块划分文件。

agent_demo/ ├── pyproject.toml 或 requirements.txt ├── config.py # 模型配置、环境变量 ├── tools/ │ ├── __init__.py │ └── system_tools.py # 自定义工具 ├── graph/ │ ├── __init__.py │ ├── state.py # 图状态定义 │ ├── nodes.py # 模型节点、工具节点 │ └── build_graph.py # 构建 LangGraph 图 └── main.py # 运行入口

学习阶段不需要把结构拆得太散,但至少要保证“模型配置”和“图构建”分开。很多初学者把所有代码写在一个文件里,等要切换模型或增加工具时,就不得不改一大片逻辑。

3.2 定义模型、工具和图状态

先把状态定义出来。LangGraph 的状态是连接所有节点信息的核心对象,每一轮迭代都会更新它。

# graph/state.py from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]

这里的关键是add_messages这个归约器。普通字典赋值会直接覆盖旧值,而使用add_messages后,新产生的消息会自动追加到消息列表。这样图里每个节点都能看到完整的对话历史,模型不会丢失前文。

接下来定义工具和模型节点。模型节点负责读取状态里的消息,生成新的回复;工具节点负责执行模型要求的工具调用。

# tools/system_tools.py from datetime import datetime from langchain_core.tools import tool @tool def get_current_time() -> str: """返回当前系统时间,适合回答与时间相关的问题。""" return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# graph/nodes.py from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolNode system_prompt = "你是一个擅长使用工具回答问题的助手。" def create_model_node(model_name: str): llm = ChatOpenAI(model=model_name, temperature=0) tools = [get_current_time] llm_with_tools = llm.bind_tools(tools) def model_node(state: AgentState): result = llm_with_tools.invoke( [{"role": "system", "content": system_prompt}] + state["messages"] ) return {"messages": [result]} return model_node tool_node = ToolNode(tools=[get_current_time])

ToolNode是 LangGraph 预置的工具执行节点,它会从模型返回的tool_calls中读取要执行的工具名和参数,执行后返回ToolMessage。它省去了手写“解析 tool_calls -> 执行工具 -> 包装结果”的重复代码。

3.3 构建执行流程并加入人类审核分支

有了模型节点和工具节点,接下来构建图。LangGraph 用StateGraph定义节点,用compile生成可执行对象,条件边负责决定下一跳走向哪里。

# graph/build_graph.py from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import tools_condition from graph.state import AgentState from graph.nodes import create_model_node, tool_node def build_agent(): graph = StateGraph(AgentState) graph.add_node("model", create_model_node("gpt-4o-mini")) graph.add_node("tools", tool_node) graph.add_edge(START, "model") graph.add_conditional_edges( "model", tools_condition, { "tools": "tools", END: END } ) graph.add_edge("tools", "model") return graph.compile()

这里tools_condition是一个内置判断函数:如果模型返回的tool_calls非空,就进入tools节点;否则直接进入ENDtools节点执行完后回到model,模型再次判断是否还需要继续调用工具。这样一个动态循环就形成了。

如果希望在工具执行前加入人工审核,可以在modeltools之间插入一个checkpoint节点,先暂停图执行并等待人工确认,确认后再把状态喂给tools节点。生产场景中这一步非常重要,尤其是工具涉及修改数据、发送消息、删除资源时。

graph.add_node("human_review", human_review_node) graph.add_edge("model", "human_review") graph.add_conditional_edges( "human_review", should_run_tools, { "tools": "tools", "end": END, } )

注意:人工审核节点不是把执行结果打印到控制台那么简单。真正的中断机制需要用 LangGraph 的异步检查点和持久化存储,让进程重启后仍然能恢复同一个会话状态。学习阶段可以先打印审核信息,生产环境再接入数据库或 Redis 检查点。

4. 接入 MCP:让 Agent 通过标准协议调用外部工具

4.1 MCP 协议的工作方式

MCP 的设计思路类似“软件总线”。一个 MCP Server 对外暴露若干工具,比如查询数据库、读取文件、调用内部接口;一个 MCP Client 负责连接 Server 并获取工具列表;Agent 运行时通过 client 拿到这些工具,把它们当成普通 LangChain 工具一样绑定给模型。

MCP 使用 JSON-RPC 2.0 格式传输请求和响应。工具调用链路由三部分组成:

  1. 初始化阶段:Client 与 Server 建立连接,交换协议版本和能力。
  2. 发现阶段:Client 列出 Server 暴露的工具列表,包含工具描述和参数 schema。
  3. 调用阶段:Client 发起工具调用请求,Server 执行并返回结构化结果。

这样做的好处是工具可以被多个 Agent 项目复用。同一个内部数据查询服务,可以暴露成 MCP Server,被不同团队、不同框架的 Agent 调用,而不用每个项目都写一套 HTTP SDK。

4.2 本地 MCP Server 最小 Demo

下面用一个 Python 文件演示如何快速启动一个 MCP Server,里面提供一个天气查询工具和当前时间工具。这里使用 MCP SDK 的 Server 类。

# mcp_demo_server.py import json from datetime import datetime from mcp.server.fastmcp import FastMCP mcp = FastMCP("DemoTools") @mcp.tool() def get_weather(city: str) -> str: """返回指定城市的模拟天气信息。""" # 生产环境应调用真实天气服务 result = { "city": city, "weather": "晴", "temperature": 26, "humidity": 0.4, } return json.dumps(result, ensure_ascii=False) @mcp.tool() def get_current_time() -> str: """返回当前时间,适合回答时间相关的问题。""" return datetime.now().strftime("%Y-%m-%d %H:%M:%S") if __name__ == "__main__": mcp.run(transport="stdio")

启动方式:

python mcp_demo_server.py

使用stdiotransport 时,MCP Client 会以子进程方式启动这个脚本,然后通过标准输入输出来通讯。这是学习阶段最简单的连接方式。如果需要跨机器访问,可以改用 SSE 或 Streamable HTTP 传输方式。

4.3 在 LangChain 中通过 MCP 适配器调用工具

LangChain 社区提供了langchain-mcp-adapters来把 MCP 工具转换成 LangChain 可识别的工具。下面是一个完整示例,通过MultiServerMCPClient连接上述本地 server,再交给 LangGraph Agent 使用。

import asyncio from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_openai import ChatOpenAI from langgraph.prebuilt import create_react_agent async def main(): async with MultiServerMCPClient( { "demo": { "command": "python", "args": ["mcp_demo_server.py"], "transport": "stdio", } } ) as client: tools = client.get_tools() model = ChatOpenAI(model="gpt-4o-mini", temperature=0) agent = create_react_agent(model, tools) result = await agent.ainvoke( {"messages": [{"role": "user", "content": "现在北京时间几点?"}]} ) for message in result["messages"]: print(type(message).__name__) if hasattr(message, "content"): print(message.content) if __name__ == "__main__": asyncio.run(main())

client.get_tools()会自动和本地 server 握手,把 MCP 工具转换成带 schema 的 LangChain 工具。之后你可以像使用普通工具一样把 tools 传入create_react_agent。这里我用的是create_react_agent预置 Agent,它内部已经用 LangGraph 构建好了 ReAct 循环,适合快速验证 MCP 工具是否能被正常调用。

如果要把 MCP 工具手工接入自定义 LangGraph 图,也可以这样做:

from langchain.tools import convert_to_langchain_tool langchain_tools = [] for mcp_tool in tools: langchain_tools.append(convert_to_langchain_tool(mcp_tool))

然后把langchain_tools传入ToolNode即可。区别在于,自定义图可以控制每个节点做什么,而create_react_agent更适合快速搭建标准 ReAct 流程。

5. 运行结果与验证方式

5.1 预期输出

运行上面的 MCP 案例后,正常情况会看到模型识别出“当前时间”这个需求,然后调用 MCP server 提供的get_current_time工具,最后根据工具返回结果给出最终答复。

预期输出大致是:

HumanMessage 现在北京时间几点? AIMessage AIMessage ToolMessage {"current_time": "2026-01-15 10:32:41", "timezone": "Asia/Shanghai"} AIMessage 当前北京时间是 2026 年 1 月 15 日 10:32:41。

如果模型没有决定调用工具,而是直接回答,可能需要检查模型 name 是否正确、工具描述是否清晰、系统提示词是否允许模型使用工具。

5.2 日志和状态检查

在自定义 LangGraph 图里,建议每隔一段时间打印状态快照,观察所有关键字段的变化。

def debug_node(state): print("--- state debug ---") print(state["messages"][-2:]) return {}

debug_node插入到模型节点和工具节点之间,就能看到模型是否发起了工具调用、工具调用参数是否正确、返回值是否被正确写入状态。生产环境应把这种日志接进结构化日志系统,而不是直接print

还可以在最终结果里检查完整消息记录:

for msg in result["messages"]: print(msg.type, msg.content, getattr(msg, "tool_calls", None))

这样能快速识别是哪一步丢失了信息。常见的情况是:模型发起了tool_calls,但ToolNode没有正确执行,原因通常是工具名不匹配。

6. 常见问题排查

6.1 按现象区分排查方向

以下是学习过程中最常遇到的几类问题。

问题现象常见原因检查方式处理建议
模型从不调用工具未执行bind_tools或工具描述不清打印model是否绑定了 tools调用bind_tools(tools),工具描述写清楚输入参数和使用场景
报错Tool name not found工具在模型绑定前被重命名,或工具注册顺序不一致打印所有工具名,比较模型传入的name使用@tool时显式指定name,建议统一全小写
MCP Server 启动失败子进程命令路径错误、工作目录不对手动运行python mcp_demo_server.py在 MCP 配置里写绝对路径,先手动验证 server 能跑通
LangGraph 状态被覆盖节点返回的 messages 没有用add_messages归约器检查AgentState定义使用Annotated[list, add_messages]
出现Agent execution terminated due to error.工具执行内部抛异常,或被工具节点捕获后终止查看完整异常栈或回调日志在工具函数内捕获业务异常,返回错误信息字符串,不要让异常直接抛出
模型返回参数格式校验失败工具参数 schema 和实际传入类型不匹配检查 plugin 或函数签名给工具参数写默认值,或使用UnionOptional明确可空字段

在中文技术社区里,Agent execution terminated due to error.这个报错经常出现在 LangGraph 的 agent 运行日志中。它本身只是顶层提示,真正的根因在错误上游。排查时不要只看这条信息,而是打开回调里的完整 traceback。

6.2 推荐排查链路

遇到 Agent 相关问题时,按下面顺序排查,能快速缩小范围:

  1. 先确认模型账号、API key、模型 name 是否有效,先跑一次不带工具的普通对话。
  2. 再确认工具能否被独立调用,用tool.invoke({...})直接测试。
  3. 然后确认模型是否产生了tool_calls。打印response.tool_calls
  4. 再检查ToolNode是否成功执行,看是否有对应ToolMessage返回。
  5. 最后看状态里的 messages 是否被正确归约,是否出现消息堆积或丢失。
  6. 如果接入 MCP,最后检查 MCP 连接层。先手动运行 server,再用一个独立 Python 脚本只做client.get_tools(),不接 Agent。

这套链路把问题从“Agent 整体失败”拆成“模型层、工具层、状态层、连接层”四个独立环节,每层都能单独验证。

7. 学习环境与生产环境的工程差异

7.1 学习环境怎么快速跑通

学习阶段的目标是理解概念,因此建议保持最小依赖。只安装langchainlangchain-openailanggraphlangchain-mcp-adaptersmcp几个核心包即可。工具用模拟数据,MCP server 用本地 stdio 方式,日志直接用print

学习阶段适合做的:

  • 所有工具函数先写返回写死的模拟数据。
  • Agent 循环最多控制在 3 到 5 轮,加一个recursion_limit参数防止死循环。
  • 先跑内置工具,再接入 MCP,分两步降低排错难度。
  • 每改动一个功能,就运行一次极简案例验证。

7.2 上生产前需要补齐的能力

生产环境的 Agent 和 demo 的最大差异不在模型能力,而在稳定性、可观测性和安全边界。

方面学习环境生产环境
模型配置代码里写死配置中心或环境变量管理
工具数据模拟数据真实数据源,需鉴权和限流
日志print结构化日志,记录每个工具调用耗时
状态存储内存Redis、PostgreSQL 等持久化检查点
异常处理让异常抛出捕获异常并转成可读结果,写入错误信息
人工审核高风险工具前加入审批节点
监控统计成功率、调用次数、token 消耗、延迟

其中最少被重视但又最关键的是状态持久化。LangGraph 执行过程中的messages默认保存在内存中,进程重启后状态就丢失。生产环境要把checkpointer接上。下面是一个用内存 checkpointer 的示例,理解后可以替换成数据库版本。

from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() app = graph.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "user-session-001"}} result = app.invoke({"messages": [{"role": "user", "content": "你好"}]}, config)

thread_id是用于区分不同会话的关键参数。后续再传入同一个thread_id,Agent 会继续读取之前的消息记录。生产环境应将MemorySaver替换为基于 Redis 或 Postgres 的持久化实现,这样服务重启后会话不中断。

8. 最佳实践和扩展方向

8.1 工具设计要小粒度、明确、可解释

一个工具只做一件事,工具描述要写清楚参数含义、返回格式和典型使用场景。不要设计一个名为process_request的万能工具。模型判断工具调用依赖的是工具名称和描述,描述不清会让模型选错工具。

推荐写法:

from langchain_core.tools import tool @tool def get_order_status(order_id: str) -> str: """根据订单ID查询订单当前状态,返回状态名称。订单号example: 202601150001。""" # 调用真实订单服务或查询数据库 return "PAID"

这里有三个细节:参数类型注明str,描述里给出示例值,返回内容说明格式。模型看到这些信息后,才能正确从用户文本里抽取参数。

8.2 状态管理要显式化,不要随手改全局变量

Agent 执行过程中会经过多轮工具调用,临时信息需要保存时,优先放进 state。例如查询用户信息后,可以在 state 里定义user_profile字段,后续节点直接读取,而不是靠模型上下文里带着大段 JSON。这样既能减少 token 消耗,又能让状态变化在日志里直观可见。

class AgentState(TypedDict): messages: Annotated[list, add_messages] user_profile: dict query_standalone: str

设计状态字段时注意字段职责单一,不要把临时变量堆到一个dict里,否则后一个节点很难知道哪些字段已经更新。

8.3 从单 Agent 到多 Agent 的扩展方向

当任务链路变长后,把思考、调用工具、结果整理都塞进一个 Agent 会让 Prompt 越来越长,也会让模型决策不稳定。可以考虑拆成多个 Agent,由主管 Agent 负责任务分配,由子 Agent 负责某一类具体操作。

例如客服系统可以分成三个子 Agent:

  • 订单 Agent:负责查询订单、退款申请。
  • 产品 Agent:负责商品搜索、库存查询。
  • 售后 Agent:负责售后表单、物流投诉。

LangGraph 里可以用多节点结构管理这些子 Agent 之间的协作。每个子 Agent 是一个独立图,主图负责调度。这样做的好处是单个 Agent 的逻辑可以单独测试、单独升级,故障范围也被限制在某个子图中。

再往后,可以关注 MCP 生态的发展和工具生态扩展。MCP 让第三方工具可以按统一协议接入,未来 Agent 应用会越来越像“操作系统连外部设备”:核心编排逻辑保持稳定,外部能力通过标准接口不断扩展。学习阶段建议先用一个最小闭环跑通 LangChain + LangGraph + MCP 的全链路,再逐步加入持久化、人工审核、多 Agent 调度和监控告警。真正能用在生产环境的 Agent 不是一次写出来的,而是从最小可行版本开始,通过日志和线上反馈不断收敛出来的。对你来说,最有价值的练习不是抄一个完整项目,而是能把工具设计、状态流转、异常处理和恢复机制这四个环节分别拆开又组合起来。

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

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

立即咨询