企业官网建设流程全解析

如果你正在学习大语言模型应用开发，接触过 LangChain，最近大概率会被一个词刷屏：LangGraph。各大技术社区、B 站教程、GitHub 开源项目里都开始出现它的身影。但你打开官方文档时，很快会被一堆概念搞懵：State、Node、Edge、Conditional Edge、Subgraph、RecursionLimit……英文文档读着费劲，中文资料又很多是表面翻译，真正能落地跑通的教程少之又少。

这篇文章的目标很直接：用一条清晰的实战主线，把 LangGraph 最核心的四个能力讲透。包括：如何定义状态、如何用节点组装流程、如何用条件边做分支和循环、如何用子图和并行分支应对复杂业务场景。同时，我会专门回答一个搜索量很高的问题：在节点函数里到底怎么改 State 状态值。

先说一个核心判断：LangGraph 真正降低的不是“调用大模型”的门槛，而是“编排多条工具调用、多步推理流程”的工程复杂度。如果你只是写一个“输入一句话、返回一段回答”的简单 Demo，不需要 LangGraph；但如果你要做带工具调用、多步判断、人工介入、可恢复状态的 Agent，LangGraph 是目前少数能优雅解决这些问题的框架。

1. 为什么 LangGraph 值得学：它真正解决的问题

1.1 LangChain 的痛点

LangChain 早期最常用的抽象是 Chain，也就是把 Prompt、LLM、工具按顺序串起来。这个设计在处理“固定步骤”时非常好用，比如：先查数据库，再写总结，最后翻译成英文。但真实的 Agent 场景不是固定的线性流程，而是有判断、有循环、有分叉的图结构。

举个例子：用户问“帮我查一下上海明天天气，再对比一下北京”。传统 Chain 很难优雅地表达“先判断用户提到了几个城市，再分别查询，最后合并结果”这样的流程。就算强行写，代码也会变成大段 if-else，维护成本非常高。

1.2 LangGraph 与 LangChain 的关系

LangGraph 是 LangChain 团队推出的“有状态、可编排”的 Agent 框架。它不是 LangChain 的替代品，而是底层流程引擎。你可以继续用 LangChain 的 Prompt 模板、大模型封装、工具集成，然后把这些组件放进 LangGraph 的图里，由 LangGraph 来控制执行顺序、状态流转、分支跳转和循环退出。

两者最直白的区别是：

很多朋友问“langchain 和 langgraph 的区别”，答案不是谁取代谁，而是分工不同。你把 LangChain 当工具箱，把 LangGraph 当流水线控制系统，配合使用效率最高。

1.3 读完这篇文章你能得到什么

读完并动手跑完整篇文章的示例后，你应该能完成三件事：第一，独立理解 LangGraph 官方文档中 State、Node、Conditional Edge 这些基本概念；第二，用 LangGraph 写一个带条件路由和循环的 Agent 流程；第三，知道在图里哪些地方容易踩坑，以及生产环境里推荐怎么组织代码。

2. 核心概念：State、Node、Edge 与执行原理

2.1 State 状态：Agent 的“记忆黑板”

LangGraph 的核心是 State。你可以把它理解成一张在所有节点之间流动的“黑板”。每个节点都能读取黑板上的内容，也能往黑板上写新内容，下一个节点读到的就是更新后的黑板。

State 通常通过 TypedDict 定义，例如：

from typing import TypedDict class ChatState(TypedDict): messages: list user_id: str

在简单的顺序流程里，State 就是一个字典；在复杂场景下，可以通过Annotated和 reducer 控制某个字段的更新方式，这个后面详细讲。

2.2 Node 节点：每一步只做一件事

Node 是图里的最小执行单元，本质上就是一个函数。函数输入是当前 State，输出是一个字典，字典里的键会被更新到 State 中。

def my_node(state: ChatState) -> dict: return {"user_id": "u_123"}

设计原则是“一个节点只做一件事”。不要在一个节点里既调模型又查数据库又做判断，拆开之后，流程可观测性和复用性都会好很多。

2.3 Edge 与 Conditional Edge：流程的“红绿灯”

Edge 是节点之间的连线，表达“当前节点执行完，下一步去哪个节点”。普通 Edge 是固定连接，Conditional Edge 则是“根据当前 State 动态决定下一步走向”。

Conditional Edge 是 LangGraph 最吸引人的点，也是很多教程没讲透的地方。它由一个路由函数和一张路由表组成：路由函数读 State，返回一个字符串；路由表把字符串映射到对应节点。这就是 Agent 能“判断”的关键机制。

2.4 子图、并行、循环：复杂流程的三种组合方式

学习了 Node 和 Edge 之后，图的基本单位已经足够，但真实业务还要用到三种组合方式：

• 循环：Conditional Edge 指回之前的节点，实现“做得不对就再来一次”。
• 子图：把一个完整流程封装成一个图，再作为一个节点嵌入父图。
• 并行：从一个节点同时发散到多个节点，再合并结果。

这三种能力组合起来，基本能覆盖 90% 的 Agent 编排需求。

3. 环境准备与安装

3.1 运行环境要求

LangGraph 是 Python 库，建议使用 Python 3.9 及以上版本，本文示例在 Python 3.10 环境下验证。安装前建议先创建虚拟环境，避免污染系统 Python。

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

3.2 安装 LangGraph

安装核心库即可：

pip install langgraph

本文示例涉及读取大模型，但为了降低联网和 API Key 的干扰，我会尽量让示例不依赖真实大模型。如果后面你希望接入 OpenAI 兼容接口，需要额外安装：

pip install langchain-openai python-dotenv

版本提示：LangGraph 迭代很快，API 可能有细微调整，本文示例以较新的稳定版为准。如果你安装的是老版本，建议先升级：

pip install --upgrade langgraph

3.3 验证安装

在命令行执行：

python -c "import langgraph; print(langgraph.__version__)"

如果正常输出版本号，说明安装成功。如果报ModuleNotFoundError，检查是否激活了正确的虚拟环境。

4. 第一个 LangGraph 应用：跑通一条完整流程

4.1 定义 State

先来一个最小示例：两个节点按顺序执行，分别向 State 中追加一条消息，并更新计数。

# graph_basic.py from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class BasicState(TypedDict): messages: Annotated[list, operator.add] count: int

这里messages使用了Annotated[list, operator.add]，含义是：每次节点返回messages字段时，不覆盖旧值，而是用operator.add把新旧列表拼接起来。这是 LangGraph 中 reducer 的典型用法，后面会细讲。

4.2 创建节点并连线

定义节点函数，并创建图结构：

def node_a(state: BasicState) -> dict: return { "messages": [f"A 收到 count={state['count']}"], "count": state["count"] + 1, } def node_b(state: BasicState) -> dict: return { "messages": ["B 处理完成"], "count": state["count"] + 1, } builder = StateGraph(BasicState) builder.add_node("node_a", node_a) builder.add_node("node_b", node_b) builder.add_edge(START, "node_a") builder.add_edge("node_a", "node_b") builder.add_edge("node_b", END)

START和END是 LangGraph 内置的特殊节点。START表示流程入口，END表示流程结束。

4.3 编译并运行

图构建好后，需要调用compile()得到可执行对象：

graph = builder.compile() result = graph.invoke({ "messages": [], "count": 0, }) print(result)

运行结果中，messages会是两条消息拼接后的列表，count变成 2。这证明两个节点按顺序执行，且 State 在整个流程中保持共享。

这个最小示例虽然简单，但已经包含了 LangGraph 最关键的机制：State 定义、节点函数、普通边、编译执行。后面所有复杂功能都是在这个基础上扩展的。

5. 条件路由：conditional_edge 深度解析与分支控制

5.1 为什么需要条件路由

顺序执行无法应对真实业务。用户输入“今天天气如何”，Agent 需要走天气查询工具；输入“3.5 乘以 8”，Agent 需要走计算工具。传统 if-else 写在单个函数里，会导致一个函数越来越长。LangGraph 的做法是把“走向判断”抽成一条边，让路由函数只负责决策，业务节点只负责执行。

5.2 路由函数与路由表

条件路由由两部分组成：

• 路由函数：接收 State，返回一个字符串 key。
• 路由表：字典，key 对应目标节点名。

写法如下：

builder.add_conditional_edges( "analyze", # 路由函数的源节点 route_function, # 路由函数 { "weather": "weather_node", "math": "math_node", "default": END, }, )

当analyze节点执行完后，LangGraph 会调用route_function(state)，拿到返回的字符串 key，再根据路由表跳到对应节点。

5.3 完整示例：按问题类型分发

下面的示例用规则区分“天气”和“计算”两类问题，完全不需要大模型，方便本地直接跑通：

# graph_conditional.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class RouteState(TypedDict): query: str route: str result: str def analyze(state: RouteState) -> dict: query = state["query"] if "天气" in query: return {"route": "weather"} if "计算" in query: return {"route": "math"} return {"route": "default"} def weather_node(state: RouteState) -> dict: return {"result": "今天晴转多云，适合出行。"} def math_node(state: RouteState) -> dict: # 这里只是演示，生产环境不要直接计算任意表达式 expression = state["query"].replace("计算", "").strip() return {"result": f"收到计算需求：{expression}"} def route_function(state: RouteState) -> str: return state["route"] builder = StateGraph(RouteState) builder.add_node("analyze", analyze) builder.add_node("weather_node", weather_node) builder.add_node("math_node", math_node) builder.add_edge(START, "analyze") builder.add_conditional_edges( "analyze", route_function, { "weather": "weather_node", "math": "math_node", "default": END, }, ) builder.add_edge("weather_node", END) builder.add_edge("math_node", END) graph = builder.compile() print(graph.invoke({"query": "上海明天天气"})) print(graph.invoke({"query": "计算 12345 + 6789"}))

运行后，两次 invoke 会走向不同节点，最终result字段内容不同。这就是最核心的分支控制。

5.4 循环检测与条件路由的配合

条件路由不仅能做分支，还能做循环。只要路由函数返回一个“之前的节点”，流程就会形成循环。循环时最容易出现死循环，因此 LangGraph 提供了一个安全机制recursion_limit。如果一次运行中节点执行次数超过该限制，会抛出异常。

try: graph.invoke({"query": "循环测试"}, config={"recursion_limit": 10}) except Exception as e: print("触发限制：", e)

实际开发中，这个配置要保留，防止模型判断失误导致 Agent 无限循环。

6. 循环与循环检测：让 Agent 学会“再来一次”

6.1 Agent 中的循环长什么样

Agent 的典型循环模式是“思考并行动”。大模型根据当前信息决定调用某个工具，拿到工具结果后再次思考，直到认为已经得到最终答案。每一步都可以看作图中的一个节点，而“是否继续”就是一条从“判断节点”指回“思考节点”的边。

6.2 用条件边实现循环

下面示例模拟一个最多执行 5 次迭代的流程：

# graph_loop.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class LoopState(TypedDict): step: int result: str def process(state: LoopState) -> dict: step = state["step"] + 1 if step < 5: return {"step": step, "result": f"第 {step} 次迭代"} return {"step": step, "result": "迭代完成"} def controller(state: LoopState) -> str: return "process" if state["step"] < 5 else "end_node" def end_node(state: LoopState) -> dict: return {"result": "最终结束：" + state["result"]} builder = StateGraph(LoopState) builder.add_node("process", process) builder.add_node("end_node", end_node) builder.add_edge(START, "process") builder.add_conditional_edges( "process", controller, { "process": "process", "end_node": "end_node", }, ) builder.add_edge("end_node", END) graph = builder.compile() result = graph.invoke({"step": 0, "result": ""}) print(result)

关键在于controller返回"process"时，流程会回到process节点，形成循环。这是 LangGraph 中实现“不断重试、不断执行工具”的通用方式。

6.3 防止死循环：recursion_limit 与步骤计数

在需要调用外部 API 的 Agent 场景中，模型可能反复选择同一个工具，导致流程无法结束。推荐双重保护：

第一，在invoke时传入recursion_limit：

graph.invoke(input_state, config={"recursion_limit": 20})

第二，在 State 中加入step字段，由节点函数负责增长计数，并在路由函数中判断上限。只要进入容错分支，就跳到结束节点，避免一直空转。

从 LangGraph 的行为来看，recursion_limit更像安全网，不要依赖它做业务结束判断。业务上的“是否结束”应该由路由函数显式决定。

7. 子图与并行分支：复杂流程的两种组合方式

7.1 什么是子图，什么时候用

子图就是把一个流程封装成图，再当成节点嵌入到另一个图里。当某个子流程足够独立时，就应该拆成子图。

典型例子：一个“文章自动发布”系统，主图负责“选题、写稿、审核、发布”，其中“写稿”内部又可以拆成“搜索资料、生成大纲、逐段生成、合并润色”四个步骤。这四个步骤内聚度很高，对外只暴露“输入主题、输出文章”，非常适合做成子图。

7.2 子图作为节点接入父图

子图编译后的对象可以直接作为节点函数使用。假设已经有一个编译好的writing_subgraph：

parent_builder = StateGraph(ParentState) parent_builder.add_node("topic_select", topic_select_node) parent_builder.add_node("writing", writing_subgraph) parent_builder.add_node("publish", publish_node) parent_builder.add_edge(START, "topic_select") parent_builder.add_edge("topic_select", "writing") parent_builder.add_edge("writing", "publish") parent_builder.add_edge("publish", END)

子图内部怎么跑，父图不关心；父图只关心子图的输入 State 和输出 State。这能显著降低复杂系统的认知负担。

7.3 并行分支：并发执行与结果合并

并行分支适合“多个独立任务同时执行”的场景。比如用户要同时查三个城市的天气，没必要串行查询，可以在一个节点分裂出三个分支，都指向合并节点。

LangGraph 支持从一个节点发出多条边，这些目标节点会在同一个 super-step 中并行执行。示例：

# graph_parallel.py from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class ParallelState(TypedDict): cities: list results: Annotated[list, operator.add] done: bool def prepare(state: ParallelState) -> dict: return {"cities": ["北京", "上海", "广州"]} def query_beijing(state: ParallelState) -> dict: return {"results": ["北京：晴"]} def query_shanghai(state: ParallelState) -> dict: return {"results": ["上海：小雨"]} def query_guangzhou(state: ParallelState) -> dict: return {"results": ["广州：多云"]} def merge(state: ParallelState) -> dict: return {"done": True, "results": state["results"]} builder = StateGraph(ParallelState) builder.add_node("prepare", prepare) builder.add_node("query_beijing", query_beijing) builder.add_node("query_shanghai", query_shanghai) builder.add_node("query_guangzhou", query_guangzhou) builder.add_node("merge", merge) builder.add_edge(START, "prepare") builder.add_conditional_edges( "prepare", lambda state: ["query_beijing", "query_shanghai", "query_guangzhou"], )

注意这里add_conditional_edges的路由函数可以直接返回一组节点名，让流程同时走向多个节点。LangGraph 会尝试并行执行它们。

最后连上合并节点：

builder.add_edge("query_beijing", "merge") builder.add_edge("query_shanghai", "merge") builder.add_edge("query_guangzhou", "merge") builder.add_edge("merge", END) graph = builder.compile() result = graph.invoke({"cities": [], "results": [], "done": False}) print(result)

合并的关键是results字段使用了Annotated[list, operator.add]。如果没有这个 reducer，多个并行节点返回results时，后写的会覆盖先写的，最终只能拿到一个城市的结果。这是并行分支最容易踩的坑。

8. 在节点函数中改变 State 状态值：正确姿势与常见误区

搜索“langgraph如何在节点函数改变state状态值”的人很多，这里专门展开讲。

LangGraph 官方推荐的更新方式是：节点函数返回一个字典，字典中键与 State 中的字段对应。

def update_node(state: MyState) -> dict: return {"name": "张三", "age": 18}

执行这个节点后，State 中name变为“张三”，age变为 18。这个方式最简单、最可控，也是官方文档示例的主路径。

8.1 返回 dict 更新状态

适合覆盖式更新单个字段。注意：如果没有在 State 字段上指定 reducer，返回的新值会直接覆盖旧值。

class ProfileState(TypedDict): name: str age: int def change_name(state: ProfileState) -> dict: # 读取 current = state["name"] # 返回更新 return {"name": current + "_new"}

8.2 使用 reducer 合并状态

如果多个节点同时更新同一个字段，或者希望“追加”而不是“覆盖”，就需要 reducer。

from typing import Annotated import operator class ListState(TypedDict): history: Annotated[list, operator.add] def node_1(state: ListState) -> dict: return {"history": ["第一步"]} def node_2(state: ListState) -> dict: return {"history": ["第二步"]}

两个节点都执行后，history是["第一步", "第二步"]，而不是只剩["第二步"]。这个机制在并行分支和子图状态回传中尤其重要。

8.3 避免直接修改 State 的三个原因

很多人为了省事，会在节点函数里直接写：

state["name"] = "张三"

这样做在部分场景下能工作，但我不推荐。原因有三个：

第一，LangGraph 的状态更新机制依赖返回值，直接修改不等于显式更新，在开启持久化、时间旅行、断点功能时可能无法正确记录状态变化。第二，函数变得难以测试，因为你既要传入 State，又需要关心函数是否修改了入参，不符合纯函数习惯。第三，遇到 reducer 时，直接修改很容易绕过合并逻辑，导致数据出现意想不到的覆盖。

更稳妥的写法是：每个节点函数只通过返回 dict 声明要更新的字段。读取操作直接访问state，修改操作全部交给返回值。这样图和状态流都非常清晰。

9. 常见问题与排查思路

排查 LangGraph 问题时，最有效的工具不是反复看文档，而是把每个节点的返回值和中间 State 打印出来。你可以把graph.invoke改成逐步调试，也可以给节点函数加日志：

import logging logging.basicConfig(level=logging.INFO) def my_node(state: MyState) -> dict: logging.info("my_node 收到 state: %s", state) return {"result": "ok"}

看日志比猜逻辑快得多。

10. 最佳实践与后续学习方向

10.1 工程建议

第一，State 定义要克制。很多项目一开始就把所有临时变量都放进 State，导致图很难调试。State 只保存需要在节点之间传递的业务数据，不保存无关的中间计算结果。

第二，节点函数保持“读 State，返回新 State”的纯函数风格。不要做文件操作后不返回任何状态，也不要在一个节点里塞太多逻辑。一个节点做太多事，会导致断点续跑和错误定位都很痛苦。

第三，所有外部调用都要设置超时和异常兜底。大模型接口、数据库、第三方 API 都可能失败，失败后路由到重试节点还是结束节点，需要在画图前就想清楚。

第四，生产环境考虑使用 LangGraph 的持久化扩展。它能记录每一步状态，支持断点恢复和人工介入。对需要“用户确认后继续”的 Agent 流程，这是很有价值的能力。

第五，测试要覆盖路由函数。分支是最容易出 bug 的地方，建议单独测试路由函数在不同输入下返回的目标节点。

10.2 后续学习方向

这篇文章只覆盖了 LangGraph 的基础图结构。如果再深入，建议按这个顺序学习：

• State 的进阶用法：Pydantic State、多个 reducer 组合、状态校验。
• Checkpoint 与持久化：把图运行中间状态保存到数据库。
• 断点与人工审核：在关键节点暂停，等人工确认后再继续。
• 多 Agent 协作：多个图或子图之间传递结果。
• LangGraph Platform：把调试好的图部署成服务。

这些能力并不是并列的，而是层层递进。先把本文中的最小示例和条件路由跑通，再往持久化和部署方向走，就不会被底层概念劝退。

10.3 最后提醒

LangGraph 的学习曲线不算陡，但它和普通链式调用完全不同。不要期望一次性写出完美的大图，我的建议是从“一个节点 + 一条边”开始，逐步加条件分支，再加循环，再加子图和并行。每加一个能力，都先打印一次 State，确认状态流向符合预期，再继续下一层。

另外，有朋友问 LangGraph 是否有 Rust 版本。目前官方主推是 Python 和 TypeScript 版本，并没有看到官方维护的 Rust 版本。如果你所在团队是 Rust 技术栈，更务实的路径是通过 HTTP API 或 LangGraph Platform 调用服务端，而不是强行把图逻辑落到 Rust 代码里。

通篇读到这里，你已经有能力把 LangGraph 放进自己的 Agent 项目里了。推荐的做法是找一个真实小需求，比如“用户输入城市名，查天气并生成出行建议”，用 LangGraph 重写一遍。写的过程中遇到异常，就对照常见问题表排查，很快能积累属于自己的实战经验。