LangGraph实战：从状态图到MCP，构建可控的Agent编排流程

很多同学在学 Agent 开发时都会遇到同一个困惑LangChain 的 Agent 用起来很爽几个工具一绑就能自动规划、调用、返回结果但一旦业务变复杂要求“先判断用户意图再决定走哪个分支”“多步工具调用后需要人工审批”“两个子任务要并发执行再汇总”LangChain 默认的 Agent 执行链就变得很难控制。你既不知道下一步会调哪个工具也无法在中间插入一个判断节点。这就是我推荐你认真学一下LangGraph的原因。本文不会只讲概念我会从 LangGraph 的底层设计说起然后用三个完整案例带你把“图谱编排”这件事跑通入门示例、条件路由、子图与并行分支最后接入 MCP 工具协议让 Agent 具备动态扩展工具的能力。无论你是刚接触 LangChain 的新手还是想深入 Agent 底层编排的开发者这篇文章都值得收藏。1. 为什么需要 LangGraphAgent 编排到底难在哪先回到一个基础问题普通聊天机器人调用大模型只需要一轮 prompt但真正的 Agent 应用往往要经历“理解用户请求 - 拆解任务 - 调用工具 - 观察结果 - 决定下一步”这样循环往复的过程。LangChain 早期的 Agent 实现负责了这个循环但它把编排逻辑封装成了一个黑盒。黑盒带来的问题很直接业务上需要“如果天气接口超时就切换备用接口”这样的条件判断时你很难在 Agent 黑盒里插入一条自定义分支。你只能用复杂的 prompt 提示模型“你应该判断一下然后用什么工具”结果模型经常理解错。LangGraph 的解决思路非常朴素把 Agent 的每一次决策、每一次工具调用、每一次状态更新都建模成一张“图”。图里有节点Node节点是数据处理函数有边Edge边是节点之间的连接关系还有状态State是整个流程中共享的数据容器。你在图上显式画出“A 节点执行完 - 判断条件 - 走 B 还是 C”流程就是可控、可读、可测试、可恢复的。从工程角度看LangGraph 带来的价值还有三点可控性执行路径由代码决定而不是完全交给模型自由发挥。可恢复每一个超级步骤Super-step都会处理状态天然适合断点续跑、人工介入。可测试节点是普通函数可以单独单测也能整图集成测试。2. LangGraph、LangChain、MCP 到底是什么关系很多初学者看到 LangGraph、LangChain、MCP 三个词在一起就容易混淆。我习惯用一句话概括LangChain 是工具链LangGraph 是编排框架MCP 是工具接入协议。LangChain 提供了大量开箱即用的组件模型封装、Prompt 模板、向量库、文档加载器、输出解析器、各种工具。早期所有人都在 LangChain 上做 Agent但 LangChain 的 Agent 执行逻辑偏自动化缺少精细化编排能力。LangGraph 的定位就是 Agent 编排层。它负责控制流程什么时候调用模型、什么时候调用工具、什么时候结束。值得注意的是LangGraph 并不绑定 LangChain 的模型封装。即使你只用原生 OpenAI SDK也能用 LangGraph 来编排流程。当然结合 LangChain 会更高效因为 Chain 里的模型、工具、解析器都能直接复用。MCPModel Context Protocol模型上下文协议则是一个更底层的工具通信标准。以前的工具接入是“每家一套 SDK”现在 MCP 定义了一套统一协议让任何 LLM 应用都能像访问 USB 设备一样发现并调用 MCP Server 暴露的工具。LangGraph 可以通过适配层把 MCP 工具加载进来作为图中的工具节点这样 Agent 的工具集就不再是写死的而是由 MCP Server 动态提供。可以用一个简单的表格来看三者的边界组件核心职责典型问题LangChain模型、Prompt、工具、链的封装如何封装大模型能力LangGraph节点、边、状态、条件的编排如何控制 Agent 流程MCP工具发现与调用协议如何统一接入外部工具3. 环境准备与版本说明本文代码基于 Python 开发建议使用 Python 3.9 及以上版本。LangGraph 的版本迭代速度较快不同大版本之间 API 会有细微差异本文示例以当前常见的 0.4.x 系列 API 为准。如果你使用的是其他版本个别类名或参数可能不同核心思路不变。先用 pip 安装必要的依赖pip install langgraph langchain-core langchain-openai如果需要接入 MCP 工具还需要安装适配包pip install langchain-mcp-adapters mcp验证安装是否成功python -c from langgraph.graph import StateGraph; print(LangGraph OK)如果输出LangGraph OK说明环境没有问题。本文的示例项目结构如下langgraph-demo/ ├── basic_graph.py # 入门示例最小状态图 ├── conditional_route.py # 条件路由实战 ├── subgraph_demo.py # 子图与并行分支 └── mcp_demo.py # MCP 工具接入示例4. 核心概念State、Node、Edge 与三种边在写正式案例之前先把 LangGraph 的四个核心概念讲清楚。理解这四个概念后面的代码就是一马平川。4.1 State节点之间共享的“内存”State 是 LangGraph 中所有节点共享的数据容器通常用一个TypedDict来定义。每个节点函数接收当前的 State返回一个字典返回的字典会更新 State。这里的更新逻辑很关键默认是“直接覆盖同名字段”但你可以用Annotated和operator.add等 reducer 实现追加、合并等更复杂的效果。from typing import TypedDict class State(TypedDict): messages: list current_step: int4.2 Node图中的执行单元Node 就是一个普通函数签名统一是(state) - dict。它负责读取 State 中的数据执行业务逻辑返回更新后的片段。def call_model(state: State): # 调用模型的逻辑 return {messages: state[messages] [model output]}4.3 Edge连接节点的路径LangGraph 的边分为三种普通边无条件从一个节点走到下一个节点。条件边根据当前 State 动态决定下一步对应add_conditional_edge。汇合边多个并行分支结束后汇聚到同一个节点继续执行本质上是多条普通边指向同一个节点。这里顺便说一个新手最容易踩的坑很多人以为 State 更新等于整个 State 对象都被替换。其实每个节点只返回“增量字段”LangGraph 会把你返回的字段合并进全局 State。如果节点 A 返回了{a: 1}节点 B 返回了{b: 2}最终的 State 会同时包含a和b。5. 入门示例第一个 LangGraph 程序先写一个最小可运行的 LangGraph 程序让流程从入口节点走到第一个节点再走到第二个节点最后结束。创建basic_graph.pyfrom typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): message: str def node_a(state: State): print(进入 node_a) return {message: state[message] - A} def node_b(state: State): print(进入 node_b) return {message: state[message] - B} # 1. 创建状态图 graph StateGraph(State) # 2. 注册节点 graph.add_node(node_a, node_a) graph.add_node(node_b, node_b) # 3. 连接边 graph.add_edge(START, node_a) graph.add_edge(node_a, node_b) graph.add_edge(node_b, END) # 4. 编译图 app graph.compile() # 5. 执行图 result app.invoke({message: start}) print(最终 State, result)运行结果进入 node_a 进入 node_b 最终 State {message: start - A - B}从这个例子可以看到LangGraph 的编程范式非常统一定义 State - 注册节点 - 连接边 - 编译 - 执行。后面所有复杂流程都是在这个范式上叠加。6. 实战一条件路由与分支控制conditional_edge 深度解析假设现在要做一个小型客服机器人用户输入包含“天气”就走天气查询流程包含“订单”就走订单查询流程如果都没匹配则走兜底回复流程。这正是条件路由的经典场景。创建conditional_route.pyfrom typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): user_input: str def parse_input(state: State): # 模拟意图识别 text state[user_input] print(f收到用户输入{text}) return {user_input: text} def weather_node(state: State): result f天气查询已为你查询 {state[user_input]} 的天气当前晴天。 print(result) return {user_input: result} def order_node(state: State): result f订单查询你的订单正在配送中。 print(result) return {user_input: result} def fallback_node(state: State): result f抱歉我暂时无法理解{state[user_input]} print(result) return {user_input: result} def route_by_intent(state: State) - str: if 天气 in state[user_input]: return weather if 订单 in state[user_input]: return order return fallback graph StateGraph(State) graph.add_node(parse, parse_input) graph.add_node(weather, weather_node) graph.add_node(order, order_node) graph.add_node(fallback, fallback_node) graph.add_edge(START, parse) # 条件边从 parse 节点出发根据 route_by_intent 的返回值路由 graph.add_conditional_edge( parse, route_by_intent, { weather: weather, order: order, fallback: fallback, }, ) graph.add_edge(weather, END) graph.add_edge(order, END) graph.add_edge(fallback, END) app graph.compile() app.invoke({user_input: 帮我查一下今天天气})运行这段代码流程会进入weather_node。如果把输入改成“我的订单到哪了”则进入order_node。这就是条件路由的核心add_conditional_edge的第二个参数是路由函数它接收当前 State返回一个字符串第三个参数字典把这个字符串映射到具体节点名。这里有两个细节需要注意。第一路由函数本身也可以直接返回目标节点的名字。LangGraph 允许省略映射字典路由函数返回什么就走哪个节点。但为了可读性和防手误我建议保留显式映射。第二条件边并不只是“二选一”一个路由函数可以返回多个目标配合后面讲的并行分支就能实现“一个节点触发多个子任务”。7. 实战二子图Subgraph与并行分支真实业务中一个流程往往有成百上千个节点。全部平铺在一张图里可维护性会很差。LangGraph 允许你把一张编译好的图嵌入另一张图作为其中的一个节点这就是子图。7.1 子图把复杂流程封装成一个节点创建subgraph_demo.pyfrom typing import TypedDict from langgraph.graph import StateGraph, START, END class SubState(TypedDict): value: int def sub_double(state: SubState): return {value: state[value] * 2} # 构建子图 sub_graph StateGraph(SubState) sub_graph.add_node(double, sub_double) sub_graph.add_edge(START, double) sub_graph.add_edge(double, END) compiled_sub_graph sub_graph.compile() class MainState(TypedDict): value: int def add_one(state: MainState): return {value: state[value] 1} # 主图 main_graph StateGraph(MainState) main_graph.add_node(add_one, add_one) main_graph.add_node(subgraph, compiled_sub_graph) main_graph.add_edge(START, add_one) main_graph.add_edge(add_one, subgraph) main_graph.add_edge(subgraph, END) app main_graph.compile() result app.invoke({value: 1}) print(result)运行结果{value: 4}流程是初始值 1 -add_one变成 2 - 子图double把 2 乘以 2 变成 4。子图完全可以当成一个普通节点使用。需要注意的是子图的 State 定义要和主图兼容否则传参时会因为缺少键而报错。7.2 并行分支Fan-out 与 Fan-in再来看一个更进阶的场景用户输入一个问题我们需要同时调用两个检索源把结果合并后再返回给模型。这里就需要并行分支。在 LangGraph 中实现并行非常直接从一个节点引出多条边分别指向不同节点。这些节点会并行执行。并行分支最终再汇聚到同一个节点。由于多个分支会同时往 State 中写数据我们最好给每个分支分配不同的键或者使用 reducer 做列表合并。from typing import TypedDict, Annotated from operator import add from langgraph.graph import StateGraph, START, END class State(TypedDict): query: str results: Annotated[list, add] # 使用 reducer 实现追加 def dispatch(state: State): print(开始分发并行任务) return {query: state[query]} def search_web(state: State): return {results: [f网页搜索结果{state[query]}]} def search_db(state: State): return {results: [f数据库结果{state[query]}]} def merge_result(state: State): print(合并后的全部结果, state[results]) return {results: state[results]} graph StateGraph(State) graph.add_node(dispatch, dispatch) graph.add_node(web, search_web) graph.add_node(db, search_db) graph.add_node(merge, merge_result) graph.add_edge(START, dispatch) # 并行分发 graph.add_edge(dispatch, web) graph.add_edge(dispatch, db) # 汇聚 graph.add_edge(web, merge) graph.add_edge(db, merge) graph.add_edge(merge, END) app graph.compile() result app.invoke({query: 什么是LangGraph, results: []}) print(result)这里的关键是Annotated[list, add]。它告诉 LangGraph当多个节点同时返回results字段时不要互相覆盖而是用operator.add把它们拼接成一个列表。如果你不加 reducer两个并行节点同时写同一个键LangGraph 会抛出状态更新冲突的异常。8. 让 Agent 接入 MCP 工具前面讲了图的编排接下来解决工具接入问题。MCPModel Context Protocol的价值在于它让工具不再依赖特定框架。你写一个 MCP Server就能在任何支持 MCP 的客户端里被调用。LangGraph 接 MCP 的常见做法是利用langchain-mcp-adapters。它可以把 MCP Server 暴露的工具转换成 LangChain 工具然后直接绑定给 LangGraph 中的模型节点使用。下面是一个配置示意import asyncio from langchain_mcp_adapters.client import MultiServerMCPClient async def build_tools(): client MultiServerMCPClient( { math-server: { url: http://localhost:8000/mcp, transport: sse, }, file-server: { command: python, args: [file_server.py], transport: stdio, }, } ) async with client as mcp_client: tools await mcp_client.get_tools() return tools if __name__ __main__: tool_list asyncio.run(build_tools()) for t in tool_list: print(工具名称, t.name)这段代码演示了两种传输方式SSE 对应远程 HTTP 服务stdio 对应本地子进程服务。如果你本地的 MCP Server 启动在别的端口或命令按需替换即可。这里要特别提醒langchain-mcp-adapters的 API 仍在快速演进不同版本方法名可能不同请以你安装版本的官方文档为准。引入 MCP 后LangGraph 的 Agent 玩法就完全变了。以前工具列表需要在代码里写死每加一个工具就要发布一次版本。现在只要 MCP Server 新增了工具Agent 在运行时就能动态发现并调用。这也是为什么 MCP 被很多人看成 Agent 生态的“统一插座”。安全性方面要强调一点MCP 工具本质上是让模型拥有了执行能力。如果你接入了一个包含“执行命令”或“删除文件”权限的 MCP Server模型有可能在误判之下触发危险操作。所以在生产环境中务必遵循最小权限原则控制 MCP Server 暴露的工具范围并对关键操作增加人工审批节点。9. 常见问题与排查思路LangGraph 的报错信息虽然清晰但新手还是会踩各种坑。我整理了几类最常见的按错误现象、原因和解决思路列出来。问题现象常见原因解决思路Invalid node input或 State 缺少某个键两个子图 State schema 不兼容检查子图和主图的TypedDict定义确保键一致并行节点写同一个字段报冲突没有使用 reducer多个分支同时返回相同键用Annotated[list, add]做合并或让分支写不同键条件路由走到了错误的节点路由函数返回值和映射字典 key 不一致先单独测试路由函数确认返回值在字典里存在调用 MCP 工具连接失败传输方式配置错误Server 未启动用 MCP 官方客户端单独测试服务确认地址和 transport图编译通过但运行不结束图中存在环但没有退出条件检查条件边是否在某一状态下能返回 END节点函数返回None导致报错节点函数没有 return 字典所有节点函数必须返回 dict至少返回空字典{}排查 LangGraph 问题时建议先小步验证把图缩小到一个节点确认能跑通再逐步加边、加分支。不要一上来就堆几十个节点那样排错成本会很高。10. 最佳实践与工程建议最后聊一些工程实现时的建议这些经验能帮你少走弯路。第一状态的字段要克制。State 里不要塞太多无关数据。每次节点返回的字段最好只包含“对后续流程有影响”的数据否则你不仅难以调试还容易在并行分支中引发冲突。第二节点函数保持纯净。我建议把“业务逻辑”和“状态更新”分开。节点函数内部尽量只做一件事读取计算返回。副作用操作如写数据库、调用外部接口放到独立的封装函数里这样单元测试时可以直接 mock。第三条件路由函数要单独测试。条件路由是 LangGraph 流程里最容易出错的点也是最值得测试的点。把路由函数抽成纯函数喂几组典型输入断言返回值符合预期再接入图中。第四善用拦截器和回调。LangGraph 支持在执行时注册回调实现日志、监控、审计。生产环境的 Agent 一定要有完整的日志链路否则模型调用了哪些工具、为什么选择了某条路径你都无从排查。第五MCP 工具接入要设置权限边界。不要把所有 MCP Server 的工具都无差别绑定给 Agent。按业务场景分角色、分组只暴露当前流程需要的工具。涉及删除、执行、支付等高危操作时建议在图中加入人工确认节点。第六版本锁定。LangGraph 和langchain-mcp-adapters更新频率非常高。项目里一定要锁定依赖版本并在升级时查看官方 changelog。我遇到过很多“昨天还能跑今天报错”的情况绝大多数都是依赖悄悄升级导致的。按照学习路径来说我建议先掌握本文的状态图、节点、边然后仿照官方文档把条件路由、子图、并行各写一遍再去尝试和 LangChain 的模型封装结合。等你对图的执行机制足够熟悉再引入 MCP 动态工具。把这个闭环跑通你已经具备了搭建生产级 Agent 的核心能力。如果这篇文章对你有帮助可以收藏备用也可以把这套代码自己改一改把节点逻辑换成你实际业务的接口跑通后再回来看看哪些地方需要调整。动手实践永远是最好的学习方式。