企业官网建设流程全解析

很多同学在学 Agent 开发时都会遇到同一个困惑：LangChain 的 Agent 用起来很爽，几个工具一绑就能自动规划、调用、返回结果；但一旦业务变复杂，要求“先判断用户意图，再决定走哪个分支”“多步工具调用后需要人工审批”“两个子任务要并发执行再汇总”，LangChain 默认的 Agent 执行链就变得很难控制。你既不知道下一步会调哪个工具，也无法在中间插入一个判断节点。这就是我推荐你认真学一下LangGraph的原因。

本文不会只讲概念，我会从 LangGraph 的底层设计说起，然后用三个完整案例带你把“图谱编排”这件事跑通：入门示例、条件路由、子图与并行分支，最后接入 MCP 工具协议，让 Agent 具备动态扩展工具的能力。无论你是刚接触 LangChain 的新手，还是想深入 Agent 底层编排的开发者，这篇文章都值得收藏。

1. 为什么需要 LangGraph：Agent 编排到底难在哪

先回到一个基础问题：普通聊天机器人调用大模型只需要一轮 prompt，但真正的 Agent 应用往往要经历“理解用户请求 -> 拆解任务 -> 调用工具 -> 观察结果 -> 决定下一步”这样循环往复的过程。LangChain 早期的 Agent 实现负责了这个循环，但它把编排逻辑封装成了一个黑盒。

黑盒带来的问题很直接：业务上需要“如果天气接口超时，就切换备用接口”这样的条件判断时，你很难在 Agent 黑盒里插入一条自定义分支。你只能用复杂的 prompt 提示模型“你应该判断一下然后用什么工具”，结果模型经常理解错。

LangGraph 的解决思路非常朴素：把 Agent 的每一次决策、每一次工具调用、每一次状态更新都建模成一张“图”。图里有节点（Node），节点是数据处理函数；有边（Edge），边是节点之间的连接关系；还有状态（State），是整个流程中共享的数据容器。你在图上显式画出“A 节点执行完 -> 判断条件 -> 走 B 还是 C”，流程就是可控、可读、可测试、可恢复的。

从工程角度看，LangGraph 带来的价值还有三点：

1. 可控性：执行路径由代码决定，而不是完全交给模型自由发挥。
2. 可恢复：每一个超级步骤（Super-step）都会处理状态，天然适合断点续跑、人工介入。
3. 可测试：节点是普通函数，可以单独单测，也能整图集成测试。

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 里的模型、工具、解析器都能直接复用。

MCP（Model Context Protocol，模型上下文协议）则是一个更底层的工具通信标准。以前的工具接入是“每家一套 SDK”，现在 MCP 定义了一套统一协议，让任何 LLM 应用都能像访问 USB 设备一样发现并调用 MCP Server 暴露的工具。LangGraph 可以通过适配层把 MCP 工具加载进来作为图中的工具节点，这样 Agent 的工具集就不再是写死的，而是由 MCP Server 动态提供。

可以用一个简单的表格来看三者的边界：

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: int

4.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.py：

from 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.py：

from 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.py：

from 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 工具

前面讲了图的编排，接下来解决工具接入问题。MCP（Model 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 的报错信息虽然清晰，但新手还是会踩各种坑。我整理了几类最常见的，按错误现象、原因和解决思路列出来。

排查 LangGraph 问题时，建议先小步验证：把图缩小到一个节点，确认能跑通，再逐步加边、加分支。不要一上来就堆几十个节点，那样排错成本会很高。

10. 最佳实践与工程建议

最后聊一些工程实现时的建议，这些经验能帮你少走弯路。

第一，状态的字段要克制。State 里不要塞太多无关数据。每次节点返回的字段最好只包含“对后续流程有影响”的数据，否则你不仅难以调试，还容易在并行分支中引发冲突。

第二，节点函数保持纯净。我建议把“业务逻辑”和“状态更新”分开。节点函数内部尽量只做一件事：读取，计算，返回。副作用操作（如写数据库、调用外部接口）放到独立的封装函数里，这样单元测试时可以直接 mock。

第三，条件路由函数要单独测试。条件路由是 LangGraph 流程里最容易出错的点，也是最值得测试的点。把路由函数抽成纯函数，喂几组典型输入，断言返回值符合预期，再接入图中。

第四，善用拦截器和回调。LangGraph 支持在执行时注册回调，实现日志、监控、审计。生产环境的 Agent 一定要有完整的日志链路，否则模型调用了哪些工具、为什么选择了某条路径，你都无从排查。

第五，MCP 工具接入要设置权限边界。不要把所有 MCP Server 的工具都无差别绑定给 Agent。按业务场景分角色、分组，只暴露当前流程需要的工具。涉及删除、执行、支付等高危操作时，建议在图中加入人工确认节点。

第六，版本锁定。LangGraph 和langchain-mcp-adapters更新频率非常高。项目里一定要锁定依赖版本，并在升级时查看官方 changelog。我遇到过很多“昨天还能跑，今天报错”的情况，绝大多数都是依赖悄悄升级导致的。

按照学习路径来说，我建议先掌握本文的状态图、节点、边，然后仿照官方文档把条件路由、子图、并行各写一遍，再去尝试和 LangChain 的模型封装结合。等你对图的执行机制足够熟悉，再引入 MCP 动态工具。把这个闭环跑通，你已经具备了搭建生产级 Agent 的核心能力。

如果这篇文章对你有帮助，可以收藏备用，也可以把这套代码自己改一改，把节点逻辑换成你实际业务的接口，跑通后再回来看看哪些地方需要调整。动手实践永远是最好的学习方式。