LangGraph入门：从LangChain到MCP的智能体流程编排实战

LangGraph 在 AI 大模型智能体开发中已经成为一个绕不开的框架。它不是一个新的编程语言而是 LangChain 生态中专门用来编排智能体流程、管理运行状态、控制分支和循环的图执行框架。很多开发者入门 LangChain 后发现链式调用可以解决普通问答、检索增强和提示词组装但一旦进入真实智能体场景——需要多轮调用工具、根据中间结果决定是否重试、多个子任务并行执行、最后还要把结果汇总起来——链式结构就不再直观。LangGraph 把这类控制流问题收敛成一张图节点负责执行业务逻辑边负责定义流转方向条件边负责动态决定下一步整个运行过程共享一份可读写的状态。这样写出来的智能体流程清楚、状态可控出问题时也可以快速定位到具体节点。本文从 LangGraph 零基础开始先说明它和 LangChain、MCP 之间的关系再准备环境并安装依赖然后通过一个最小智能体跑通完整流程之后深入条件路由、循环、并行分支、子图和 MCP 工具接入。每个部分都会给出可以直接运行的示例代码、运行方式和预期结果最后集中列出常见问题的排查方法。跟着完整示例走一遍之后你应该能自己搭建一个带条件路由、工具调用和子图复用的智能体项目并且在遇到图不执行、状态被覆盖、MCP 工具加载失败这类问题时知道从哪一层开始排查。1. 先理解 LangGraph、LangChain 和 MCP 之间的关系1.1 LangChain 已经把工具组装起来为什么还需要 LangGraphLangChain 的核心抽象是 Chain也就是一条预先定义好的调用链。比如一个典型的 RAG 流程可以写成“加载文档 - 切分 - 向量化 - 检索 - 拼提示词 - 调用模型”。这个链路在执行顺序上是固定的分支逻辑需要写在函数内部循环逻辑则需要手动用 while 或 for 去控制。但真实智能体不一样。真实智能体的下一步往往取决于上一步的执行结果。例如用户问“帮我查一下北京今天的天气如果下雨就提醒我带伞”模型先判断出需要调用天气工具工具返回后模型还要判断是否满足结束条件不满足就要继续调用其他工具。这个过程天然是一个带循环和条件判断的图而不是一条直线。LangGraph 就是为了解决这个问题出现的。它允许把流程拆成节点再把节点之间的流转关系用边表示。节点可以执行任意代码包括调用大模型、调用工具、读写数据库、调用其他服务。边可以是有向静态的也可以是条件动态的。这样整个智能体就是一个可以真正运行的控制流图。1.2 LangGraph 与 LangChain 的区别和分工LangChain 本身不负责图的执行它提供的是模型接入、消息结构、提示模板、工具定义、向量库、记忆等基础组件。LangGraph 则更聚焦在“流程”上负责决定每个节点何时执行、状态如何传递、分支如何选择、循环如何终止。实际开发中两者是配合使用的LangChain 负责大模型调用和工具封装LangGraph 负责流程编排。一个智能体节点内部可以调用 ChatOpenAI、可以调用 ConversationBufferMemory也可以调用任意 LangChain 工具但“下一步调用哪个节点”由 LangGraph 的边和条件边决定。维度LangChainLangGraph核心抽象Chain顺序执行链路Graph节点和边组成的有向图分支能力需要在函数内部手动判断通过条件边动态选路循环能力需要外部循环控制图本身允许存在循环状态管理依赖外部内存组件内置 State 全局状态适合场景问答、RAG、固定流程多步骤、有分支、有工具调用的智能体1.3 MCP 在智能体工程中的角色MCP 全称 Model Context Protocol是一种用来统一大模型应用与外部工具、数据源之间交互方式的协议。在没有 MCP 之前每个工具都需要写一套自定义接口模型调用它时要处理不同的认证、参数格式和返回结构。MCP 把这部分标准化了模型应用通过一个统一的客户端连接 MCP Server就能发现并调用 Server 暴露的工具、资源或提示模板。在 LangGraph 工作流里MCP 解决的是“工具从哪来”的问题。LangGraph 本身不关心工具是普通 Python 函数还是 MCP 远程服务只要它最终被包装成一个可以执行的 Tool 对象就能在节点中被调用。因此MCP 和 LangGraph 不是替代关系而是接入关系。2. 环境准备版本、依赖和模型接入2.1 Python 环境与虚拟环境LangGraph 是 Python 库需要 Python 3.9 及以上版本。建议不要直接在系统 Python 里安装而是新建一个虚拟环境避免不同项目之间的依赖冲突。创建虚拟环境的命令如下python -m venv langgraph-env在 Windows 下启用langgraph-env\Scripts\activate在 macOS 或 Linux 下启用source langgraph-env/bin/activate启用后通过python --version确认版本。常见的 LangGraph 项目在 Python 3.10 和 3.11 下表现最稳定Python 3.12 也能用但要确认辅助包的兼容性。2.2 安装 LangGraph、LangChain 和 MCP 相关包最小依赖包括 langgraph、langchain-core、langchain-openai。MCP 接入还需要 mcp 和 langchain-mcp-adapters。以下命令在当前虚拟环境中执行pip install -U langgraph langchain langchain-openai pip install -U mcp langchain-mcp-adapters如果需要使用本地模型配合 Ollama 时只需安装 langchain-openai因为 Ollama 提供兼容 OpenAI 的 HTTP 接口不需要额外安装 Ollama 的 Python SDK。要注意 LangGraph 的 API 迭代速度比较快。早期版本从langgraph.graph导入StateGraph从较新的版本开始统一从langgraph.graph导入StateGraph、START、END。如果安装后出现ImportError通常不是代码问题而是版本过旧。落地项目时建议把核心依赖版本固定到具体版本号例如pip freeze requirements.txt生产环境发布前应该根据 requirements.txt 在全新环境里重新安装验证避免“本地能跑服务器上跑不起来”的问题。2.3 配置大模型 API支持远程模型和本地模型LangGraph 本身不直接调用模型它通过 LangChain 的 ChatModel 接口调用。使用 OpenAI 兼容接口的配置如下import os os.environ[OPENAI_API_KEY] 你的APIKey如果使用本地模型例如 Ollama 中启动的 qwen2.5:7b可以这样配置from langchain_openai import ChatOpenAI llm ChatOpenAI( modelqwen2.5:7b, base_urlhttp://localhost:11434/v1, api_keyollama, temperature0, )这里的base_url指向 Ollama 的兼容端点。使用本地模型的优势是数据不需要发送到第三方缺点是模型能力通常不如远程模型实际效果需要以项目验证为准。3. LangGraph 核心组件逐个拆解3.1 State整个流程共享的数据结构State 是 LangGraph 图运行时的共享数据载体。每个节点都接收当前 State执行完业务后返回一个字典LangGraph 会把返回的字段更新到 State 中。声明 State 最简单的方式是使用 TypedDictfrom typing import TypedDict class AgentState(TypedDict): messages: list intent: str answer: str这里的messages用于保存对话消息intent用于保存模型判断出的用户意图answer用于保存最终回答。节点可以只返回需要更新的部分字段没有返回的字段会保持不变。有一个关键细节默认情况下节点返回的字段是“覆盖”式更新。如果一个字段被两个并行节点同时写入后执行的节点会覆盖先执行的节点。如果需要把多个结果合并就要用 Annotated 配合 reducer 函数本文后面会专门演示。3.2 Node真正的业务执行单元Node 就是一个普通 Python 函数接收当前 State返回一个字典。例如def weather_node(state: AgentState) - dict: return {answer: 天气节点查询城市天气}节点内部可以调用 LLM、调用工具、读写数据库甚至调用另一个服务。LangGraph 不限制节点内的复杂度但建议一个节点只做一件事这样打印日志、定位问题和单测都更容易。3.3 Edge 与 Conditional Edge静态路径与动态路径Edge 表示节点之间的固定流转方向。add_edge接收两个节点名builder.add_edge(node_a, node_b)意思是 node_a 执行完之后直接进入 node_b。Conditional Edge 表示根据某个函数或规则决定下一步。它接收一个路由函数这个函数接收当前 State返回一个字符串该字符串对应节点的名称builder.add_conditional_edges( decide_node, route, { weather: weather_node, finance: finance_node, general: general_node, }, )路由函数的返回值作为 key去映射表里找对应的 valuevalue 就是要执行的节点名。这样就把“下一步走哪条路”从固定编码变成了动态决策。3.4 编译与调用图构建完成后调用compile()编译成可执行对象app builder.compile()编译后的对象支持invoke同步调用和ainvoke异步调用result app.invoke({ messages: [], intent: , answer: , })invoke传入的初始字典就是 State 的初始值。编译阶段可以理解为把图结构、节点函数、边界条件绑定成一个可执行计划。4. 从最小智能体开始意图识别与路由4.1 需求设计第一个例子做一个意图识别智能体接收用户问题。先用 LLM 判断意图意图分为 weather、finance、general。根据意图进入对应节点。节点返回回答流程结束。这是 LangGraph 中最典型的“先决策后路由”场景。它虽然简单但可以把 State、Node、Conditional Edge 全部串起来。4.2 完整代码from typing import TypedDict, Literal from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: list intent: str answer: str llm ChatOpenAI( modelgpt-4o-mini, temperature0, ) def decide_intent(state: AgentState) - dict: latest_user_message state[messages][-1].content prompt ( 判断用户意图只输出一个英文词weather、finance、general。 f用户输入{latest_user_message} ) text llm.invoke([SystemMessage(contentprompt)]).content.strip() return {intent: text} def weather_node(state: AgentState) - dict: return {answer: 天气节点这里会调用真实天气服务} def finance_node(state: AgentState) - dict: return {answer: 财经节点这里会调用股票行情服务} def general_node(state: AgentState) - dict: latest_user_message state[messages][-1].content answer llm.invoke([HumanMessage(contentlatest_user_message)]).content return {answer: answer} def route(state: AgentState) - Literal[weather_node, finance_node, general_node]: intent state[intent] if weather in intent: return weather_node if finance in intent: return finance_node return general_node builder StateGraph(AgentState) builder.add_node(decide, decide_intent) builder.add_node(weather, weather_node) builder.add_node(finance, finance_node) builder.add_node(general, general_node) builder.add_edge(START, decide) builder.add_conditional_edges( decide, route, { weather_node: weather, finance_node: finance, general_node: general, }, ) builder.add_edge(weather, END) builder.add_edge(finance, END) builder.add_edge(general, END) app builder.compile()4.3 运行验证result app.invoke({ messages: [HumanMessage(content上海明天天气怎么样)], intent: , answer: , }) print(result[intent]) print(result[answer])预期输出大概是weather 天气节点这里会调用真实天气服务再换一个问题验证分支result app.invoke({ messages: [HumanMessage(content帮我看一下腾讯的股票)], intent: , answer: , }) print(result[intent]) print(result[answer])预期会走 finance 节点。4.4 关键解释decide_intent节点把模型返回的文本写入 intent 字段。route函数读取 intent返回对应的节点名。注意条件边的映射表里key 是 route 的返回值value 是图里实际注册的节点名。这个例子里的四个节点各自只做一件事跑起来后可以单独验证每个节点返回的字段。实际项目中如果发现某个意图总是走错分支先打印decide_intent返回的 intent再检查路由条件不要直接改流程。5. 进阶控制流条件路由、循环、并行分支5.1 理解 conditional_edges 的映射关系条件边的映射是 LangGraph 里最容易出错的地方。add_conditional_edges的第三个参数如果不传LangGraph 会把路由函数的返回值直接当作节点名。如果传了映射表则返回值作为 key映射表里对应的 value 才是实际要进入的节点。错误示例# route 返回 weather_node # 但图里注册的节点叫 weather builder.add_conditional_edges( decide, route, {weather_node: weather_node}, # 错value 对应错误节点名 )检查方式是打印路由函数的返回值再对照映射表。条件边的映射表不仅可以把字符串映射到节点名也可以映射到END这样分支可以终止在图的不同位置。5.2 用循环实现“调用工具 - 再判断”的 ReAct 过程智能体最常用的模式是循环LLM 判断需要调用工具工具执行后返回结果再把结果交给 LLM直到 LLM 认为可以结束。LangGraph 图允许存在环路关键是条件函数要提供明确的结束分支。先定义一个工具from langchain_core.tools import tool tool def get_current_weather(city: str) - str: 查询一个城市的天气返回天气描述。 return f{city} 今天晴天气温 22 度。再构建带循环的图from typing import Annotated, TypedDict, Literal from langchain_core.messages import HumanMessage from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode class ReActState(TypedDict): messages: Annotated[list, add_messages] llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools([get_current_weather]) tool_node ToolNode([get_current_weather]) def call_llm(state: ReActState) - dict: response llm_with_tools.invoke(state[messages]) return {messages: [response]} def should_continue(state: ReActState) - Literal[tools, end]: last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return end builder StateGraph(ReActState) builder.add_node(llm, call_llm) builder.add_node(tools, tool_node) builder.add_edge(START, llm) builder.add_conditional_edges( llm, should_continue, { tools: tools, end: END, }, ) builder.add_edge(tools, llm) agent builder.compile() result agent.invoke({ messages: [HumanMessage(content北京天气怎么样)] }) print(result[messages][-1].content)关键在于should_continue。当 LLM 返回的消息包含 tool_calls 时说明模型希望调用工具于是进入 tools 节点工具执行后返回 TOOL 消息再回到 llm 节点。当 LLM 输出不包含 tool_calls 时说明模型已经生成最终回答流程结束。这个循环是合法的LangGraph 不会阻止图结构中的环路但要求开发者自己保证最终能走到 END 分支否则可能出现无限循环。注意循环不是越少越好。ReAct 模式的每一次循环都是“执行一次工具调用再判断一次”这通常是有意义的一步。真正的问题是没有结束条件的时候LLM 会不断调用工具。生产环境建议给整个请求设置超时时间或最大递归次数。5.3 并行分支与状态合并多个互不依赖的任务可以放在并行分支里。LangGraph 支持从一个节点分发到多个节点多个节点再汇集到一个汇合节点。汇合节点会等待所有上游节点都执行完成后再运行。这里需要引入 reducer。没有 reducer 的情况下两个并行节点同时写同一个 list 字段会出现覆盖。用 Annotated 声明合并函数即可from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END def merge_results(left: list, right: list) - list: return left right class ParallelState(TypedDict): keyword: str results: Annotated[list, merge_results] answer: str def web_search_node(state: ParallelState) - dict: return {results: [f{state[keyword]} 的网页搜索结果]} def report_node(state: ParallelState) - dict: return {results: [f{state[keyword]} 的行业报告]} def combine_node(state: ParallelState) - dict: return {answer: | .join(state[results])} builder StateGraph(ParallelState) builder.add_node(web_search, web_search_node) builder.add_node(report, report_node) builder.add_node(combine, combine_node) builder.add_edge(START, web_search) builder.add_edge(START, report) builder.add_edge(web_search, combine) builder.add_edge(report, combine) builder.add_edge(combine, END) parallel_app builder.compile() result parallel_app.invoke({keyword: AI大模型, results: [], answer: }) print(result[answer])combine 节点会等 web_search 和 report 都完成后再执行。如果两个节点都想更新 results 字段merge_results 会把两个列表拼接起来而不是后写覆盖前面。并行分支是 LangGraph 对比普通 Chain 的重要能力之一但它也引入了状态竞争风险。使用并行时必须先分析哪些节点之间真正没有依赖关系不要因为“能并行”就把所有节点都改成并行。6. 子图把复用流程拆出来6.1 子图适合什么场景当一个图变得很大例如有十个节点、三条条件边、两个循环主图会很难阅读。此时可以把一段相对独立的流程封装成子图再在主图中作为一个节点挂载。子图适合以下场景一段流程被多个主图复用。某段流程内部有完整的循环或分支逻辑。单独测试一段流程比直接测试整图更方便。多人协作时按子图划分模块边界。6.2 子图的定义与挂载子图的定义方式和普通图一样最后也是调用 compile() 得到编译对象。例如把“天气详情查询”做成一棵子树from typing import TypedDict from langgraph.graph import StateGraph, START, END class WeatherSubState(TypedDict): city: str result: str def query_city(state: WeatherSubState) - dict: return {result: f{state[city]} 晴气温 26 度} sub_builder StateGraph(WeatherSubState) sub_builder.add_node(query, query_city) sub_builder.add_edge(START, query) sub_builder.add_edge(query, END) weather_subgraph sub_builder.compile()然后把这个子图挂到主图的一个节点上class MainState(TypedDict): city: str final: str def route_to_subgraph(state: MainState) - dict: return {city: state[city], result: } builder StateGraph(MainState) builder.add_node(weather_sub, route_to_subgraph) builder.add_node(weather_task, weather_subgraph)这里的关键是主图节点和子图节点之间的状态字段需要兼容。主图传给子图的字段如果子图的 State 中没有定义会无法传递造成报错。建议子图 State 只暴露最少的外部接口字段内部字段全部封装在子图内部。实际项目中子图还有一个额外好处可以单独compile()、单独测试。每次确认子图通过后再挂载到主图能显著减少排查时间。7. 接入 MCP 工具协议7.1 MCP 工具接入的基本流程MCP Server 是独立运行的服务LangGraph 端需要先建立客户端 Session然后发现工具。langchain-mcp-adapters提供了load_mcp_tools可以把 MCP 工具转换为 LangChain Tool 列表。以文件系统 MCP Server 为例先确认本机有 Node.js 环境然后启动一个标准文件服务器import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools server_params StdioServerParameters( commandnpx, args[-y, modelcontextprotocol/server-filesystem, /tmp], ) async def load_tools(): async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: tools await load_mcp_tools(session) print(f加载到 {len(tools)} 个工具) return tools tools asyncio.run(load_tools())stdio_client负责启动子进程并建立标准输入输出通道ClientSession负责 MCP 协议通信load_mcp_tools负责把 MCP 工具转换成 LangGraph 可用的 Tool 对象。7.2 用 create_react_agent 或自定义图接 MCP 工具拿到工具列表后最简单的用法是交给create_react_agentfrom langgraph.prebuilt import create_react_agent from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini) agent create_react_agent(llm, tools) result agent.invoke({ messages: [{role: user, content: 读取 /tmp 目录下的文件列表}] }) print(result[messages][-1].content)create_react_agent内部已经实现了上一节描述的 ReAct 循环不需要手动写条件边。如果业务场景更复杂也可以在自定义图里把 MCP 工具绑定到 LLM再通过 ToolNode 调用。需要留意的是MCP 工具加载是异步过程。如果项目整体是同步代码要先将异步加载的结果提前准备好再进入 LangGraph 的同步执行流程。不要在同步的invoke里直接混用 async 的 MCP Session容易出现事件循环不兼容的错误。8. 常见问题排查8.1 依赖安装与版本冲突问题现象常见原因检查方式处理建议ImportError: cannot import name STARTlanggraph 版本过旧pip show langgraph升级到 0.2 以上版本pydantic相关报错辅助包与 pydantic 版本冲突pip check固定 pydantic 版本并重新安装ModuleNotFoundError: langchain_openai未安装 langchain-openaipip show langchain-openai安装对应包处理这类问题最稳妥的方式不是逐个猜测而是新建一个干净的虚拟环境把所有依赖用 requirements.txt 一次性安装然后再跑最小示例。8.2 状态字段被覆盖和条件边映射错误问题现象常见原因检查方式处理建议并行执行后只保留一个结果field 没有 reducer打印 state 中间结果给字段加Annotated[list, merge_func]条件边总是走同一条分支route 返回值与映射 key 不一致在 route 里打印返回值对照映射表修正路由函数返回的节点名不存在节点名拼写错误检查 add_node 的注册名统一用常量管理节点名条件边排查时先隔离问题用一个固定返回值的临时路由函数跑一次如果能到目标节点说明条件函数本身的问题如果还是走错说明映射表或节点名有问题。8.3 MCP 与异步边界问题问题现象常见原因检查方式处理建议RuntimeError: asyncio.run() cannot be called from a running event loop在异步环境中直接调用 asyncio.run查看调用栈使用 await 或直接把入口设置为 async工具列表为空MCP Server 未正确启动或路径错误手动运行命令验证服务先单独测试 MCP Server 再接入 LangGraph工具调用超时子进程启动慢或网络问题增加日志观察启动耗时提高超时配置或优化启动脚本MCP 排错顺序是先确认 Server 能独立运行再确认 Session 能成功建立最后才去看工具是否被 LangGraph 正确加载。不要一上来就怀疑 LangGraph 的节点逻辑。8.4 快速排错清单确认 Python 版本和虚拟环境。用pip show查看关键包版本。打印 State 的关键字段确认节点是否真的写了目标字段。打印条件路由的返回值确认分支判断入口。用单节点独立测试代替整图调试。对 MCP Server先独立命令行启动再接入框架。生产环境记录每次运行的节点链路和耗时。9. 最佳实践与扩展方向9.1 学习环境与生产环境的差异学习环境里把依赖安装好、能跑通最小示例就够了。生产环境需要考虑更多阶段学习环境生产环境依赖直接安装最新版锁定版本并镜像依赖配置写在代码顶部或环境变量配置中心或密钥管理日志用 print 观察结构化日志记录节点和耗时状态内存态持久化或 checkpoint支持恢复工具模拟数据真实服务加上鉴权和限流异常直接抛出设置超时、重试、兜底回答监控不关注提醒、告警、指标报表9.2 可落地的实践建议节点职责单一。一个节点只做一件事不要把所有逻辑塞进一个函数。状态字段最小化。State 里只放真正需要跨节点共享的数据临时变量尽量留在节点内部。条件边返回值集中管理。用常量或枚举代替字符串散落各处避免拼写错误。先加 reducer 再合并并行结果。并行分支涉及共享字段时提前设计合并函数。循环必须有结束路径。检查 ReAct 循环里是否存在明确返回 END 的分支。为每个节点增加 trace 日志。记录入参、出参和耗时出问题时能快速定位。9.3 扩展方向LangGraph 的能力不止本文覆盖的内容。继续深入可以从这几个方向入手Checkpoint 持久化把图运行状态保存下来实现断点续跑和会话恢复。Human-in-the-loop在流程中暂停等待人工审核后再继续。多智能体协作把不同职责的智能体封装成子图在主图中按任务分配。与 LangSmith 等工具集成把图的结构、节点响应和 token 消耗完整记录下来。自定义 reducer设计叠加、取最大值、去重等复杂状态合并逻辑。LangGraph 的核心价值不是炫技而是把复杂智能体流程变成可控、可读、可测试的图结构。先跑通最小示例再逐步加入工具调用、循环和子图是理解这个框架最快的方式。遇到问题时从 State 和各节点返回值入手通常比盯着整体流程看更容易找到根因。